engineering/remarkbox
1
0
Fork 0
Code Issues 3 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge branch 'import-comments-feature' into 'main'

Browse source 
Add import comments feature with support for WordPress, Disqus, and Graphcomment

See merge request engineering/remarkbox/remarkbox!26
This commit is contained in:
Russell Ballestrini 2025-12-20 15:41:40 +00:00
commit 8b353fc7ec
8 changed files with 1255 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

10
CLAUDE.md
View file

@ -1,5 +1,15 @@
# Claude Code Configuration
## Project Setup
**IMPORTANT**: Before starting any work on a repository:
1. Check for a `CLAUDE.md` file in the repository root
2. Check for a `CLAUDE.md` file in parent directories (we often work across repos on localhost)
3. Read and follow all instructions in those files
4. These project-specific instructions override default Claude Code behavior
5. Look for conventions around commits, testing, code style, and workflows
6. If working across multiple repositories, respect the conventions from each repo's CLAUDE.md
## Commit Attribution
When creating git commits, use clean, simple commit messages:

3
remarkbox/models/namespace.py
View file

@ -107,6 +107,9 @@ class Namespace(RBase, Base):
reverse_order = Column(Boolean, default=False)
# should we group conversations and limit to nesting 2 deep?
group_conversations = Column(Boolean, default=False)
# the group postfix used for imports (e.g., "rb" creates "Anonymous-rb")
# Once set, this becomes permanent for all imported surrogates
import_group_postfix = Column(Unicode(6), default=None, nullable=True)
# the type of subscription of this Namespace.
subscription_type = Column(
Enum(*SUBSCRIPTION_TYPES, name="subscription_type"),

2
remarkbox/routes.py
View file

@ -68,6 +68,7 @@ def includeme(config):
config.add_route("embed-namespace-nodes", "/embed/ns/{namespace}/nodes")
config.add_route("embed-namespace-settings", "/embed/ns/{namespace}/settings")
config.add_route("embed-namespace-import-comments", "/embed/ns/{namespace}/import-comments")
config.add_route(
"embed-namespace-stylesheet", "/embed/ns/{namespace}/{filename}.css"
)
@ -99,6 +100,7 @@ def includeme(config):
config.add_route("basic-namespace-nodes", "/ns/{namespace}/nodes")
config.add_route("basic-namespace-settings", "/ns/{namespace}/settings")
config.add_route("basic-namespace-import-comments", "/ns/{namespace}/import-comments")
config.add_route("basic-namespace-stats-json", "/ns/{namespace}/stats.json")
config.add_route("basic-namespace-stylesheet", "/ns/{namespace}/{filename}.css")
config.add_route("basic-namespace-threads-rss", "/ns/{namespace}.threads.xml")

33
remarkbox/scripts/alembic/versions/5188e62d0afb_add_import_group_postfix_column_to_.py Normal file
View file

@ -0,0 +1,33 @@
"""Add import_group_postfix column to namespace table
Revision ID: 5188e62d0afb
Revises: b8f3c9d4e5a1
Create Date: 2025-11-24 06:39:06.299392
"""
# revision identifiers, used by Alembic.
revision = '5188e62d0afb'
down_revision = 'b8f3c9d4e5a1'
branch_labels = None
depends_on = None
from alembic import op
import sqlalchemy as sa
from sqlalchemy_utils import UUIDType as TempUUIDType
UUIDType = TempUUIDType(binary=False)
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('rb_namespace', sa.Column('import_group_postfix', sa.Unicode(length=6), nullable=True))
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('rb_namespace', 'import_group_postfix')
# ### end Alembic commands ###

83
remarkbox/templates/import-comments.j2 Normal file
View file

@ -0,0 +1,83 @@
{% extends request.base_template -%}
{% block title %}{{ the_title }} | {{ request.domain }}{%- endblock -%}
{% block content -%}
<h3>{{ the_title }}</h3>
<p>
Import comments and threads from WordPress, Disqus, or pre-formatted JSON exports using the
<a href="https://github.com/russellballestrini/blog-to-json" target="_blank" rel="nofollow">blog-to-json</a> tool.
</p>
<section class="well" style="text-align: left;">
<h4>How to Use</h4>
<ol>
<li><b>Export your comments:</b>
<ul>
<li>Install the blog-to-json tool: <code>pip install blog-to-json</code></li>
<li>Use blog-to-json to convert your export file to JSON format</li>
<li>See the <a href="https://github.com/russellballestrini/blog-to-json" target="_blank" rel="nofollow">tool documentation</a> for supported platforms and usage</li>
</ul>
</li>
<li><b>Automatic user creation:</b>
<ul>
<li>Comments with email addresses will create or match existing users</li>
<li>Comments without email addresses will create surrogate users (guest users specific to your namespace)</li>
<li>All imported users will be tagged with a unique timestamp to identify them as imported from this session</li>
</ul>
</li>
<li><b>Upload and import:</b> Select your JSON file and click "Import Comments".</li>
</ol>
<p><b>Note:</b> The import process will:
<ul>
<li>Create threads for each unique URL in your export</li>
<li>Create or find existing users by email address</li>
<li>Import all comments with their timestamps and hierarchy</li>
<li>Mark all imported comments as verified</li>
</ul>
</p>
<p><b>Get the tool:</b> <a href="https://github.com/russellballestrini/blog-to-json" target="_blank" rel="nofollow">https://github.com/russellballestrini/blog-to-json</a></p>
</section>
<br>
<form method="post" action="{{ request.link_prefix }}/ns/{{ request.namespace.name }}/import-comments" enctype="multipart/form-data" onsubmit="submit.disabled = true; submit.value = 'Importing...'; return true;">
<label>JSON File:</label>
<input type="file" name="json-file" id="json-file" accept=".json" required class="common-text-input"></input>
<br>
<small>Select the JSON file generated by blog-to-json</small>
<br>
<br>
<label for="group-prefix">Group Postfix (required):</label>
<input type="text" name="group-prefix" id="group-prefix" value="{{ default_prefix }}" class="common-text-input" placeholder="{{ default_prefix }}" minlength="2" maxlength="6" required {% if postfix_locked %}disabled{% endif %}></input>
<br>
{% if postfix_locked %}
<small><b>Locked:</b> This namespace is using the permanent postfix "<b>{{ default_prefix }}</b>". All imported users will be tagged with this postfix (e.g., "Anonymous-{{ default_prefix }}"). Re-uploading will reuse existing users with this postfix.</small>
{% else %}
<small>Short postfix (2-6 alphanumeric chars) to tag imported users. Auto-generated from your namespace domain, but you can customize it. <b>Once used, this postfix becomes permanent.</b> Re-uploading with the same postfix will reuse existing users.</small>
{% endif %}
<br>
<br>
{% include 'snippets/csrf.j2' %}
<br>
{% set submit_button_classes = 'button-right green-button' %}
{% set submit_button_value = 'Import Comments' %}
{% include 'snippets/submit.j2' %}
</form>
<br>
<br>
<a href="{{ request.link_prefix }}/ns/{{ request.namespace.name }}/settings">Back to Namespace Settings</a>
{%- endblock -%}

14
remarkbox/templates/namespace-settings.j2
View file

@ -275,4 +275,18 @@ These people may modify Namespace settings (this page).
</li>
</ul>
<br>
<hr>
<br>
<h4>Import Comments</h4>
<label>Import Comments:</label>
Import comments and threads using the <a href="https://github.com/russellballestrini/blog-to-json" target="_blank" rel="nofollow">blog-to-json</a> tool.
<br>
<br>
<a href="{{ request.link_prefix }}/ns/{{ request.namespace.name }}/import-comments" class="button">Import Comments</a>
{%- endblock -%}

834
remarkbox/tests/test_import_comments.py Normal file
View file

@ -0,0 +1,834 @@
import transaction
import unittest
import webtest
import json
import io
from remarkbox.models import (
Node,
get_tm_session,
get_or_create_user_by_email,
get_user_by_email,
get_or_create_namespace,
)
from remarkbox.models.meta import Base
from pyramid.paster import get_appsettings
try:
unicode("")
except:
from six import u as unicode
class ImportCommentsFunctionalTests(unittest.TestCase):
"""Tests for the import comments functionality"""
@classmethod
def setUpClass(cls):
from remarkbox import main
cls.settings = get_appsettings("test.ini")
cls.app = main({}, **cls.settings)
cls.testapp = webtest.TestApp(cls.app)
cls.session_factory = cls.app.registry["dbsession_factory"]
cls.engine = cls.session_factory.kw["bind"]
Base.metadata.create_all(bind=cls.engine)
cls.tm = transaction.manager
cls.dbsession = get_tm_session(cls.session_factory, cls.tm)
@classmethod
def tearDownClass(cls):
cls.dbsession.close()
Base.metadata.drop_all(bind=cls.engine)
def setUp(self):
# Create test user and authenticate
self.test_user = get_or_create_user_by_email(
self.dbsession, "owner@example.com"
)
self.raw_otp = self.test_user.new_password()
self.dbsession.add(self.test_user)
self.dbsession.flush()
self.tm.commit()
# Requery to avoid detached instance error
self.test_user = get_or_create_user_by_email(
self.dbsession, "owner@example.com"
)
# Log in the test user
self.testapp.post(
f"/verification-challenge?email=owner@example.com&raw-otp={self.raw_otp}&submit"
)
# Get CSRF token
res_csrf = self.testapp.get("/")
self.csrf = res_csrf.form.fields["csrf_token"][0].value
# Create a test namespace and make the user an owner
self.test_namespace = get_or_create_namespace(
self.dbsession, "test.example.com"
)
self.test_namespace.set_role_for_user(self.test_user, role="owner")
self.dbsession.add(self.test_namespace)
self.dbsession.flush()
self.tm.commit()
def tearDown(self):
# Clean up: log out
self.testapp.get("/log-out")
def get_test_namespace(self):
"""Helper to get fresh namespace object from DB"""
from remarkbox.models import get_namespace_by_name
return get_namespace_by_name(self.dbsession, "test.example.com")
def test_import_page_requires_authentication(self):
"""Test that the import page requires user authentication"""
self.testapp.get("/log-out")
redirect_res = self.testapp.get(
"/ns/test.example.com/import-comments", status=302
)
res = redirect_res.follow()
self.assertIn(b"You must log in to access that area.", res.body)
def test_import_page_requires_namespace_ownership(self):
"""Test that the import page requires namespace ownership"""
# Create a different user
other_user = get_or_create_user_by_email(
self.dbsession, "other@example.com"
)
other_otp = other_user.new_password()
self.dbsession.add(other_user)
self.dbsession.flush()
self.tm.commit()
# Log out current user and log in as other user
self.testapp.get("/log-out")
self.testapp.post(
f"/verification-challenge?email=other@example.com&raw-otp={other_otp}&submit"
)
# Try to access import page
redirect_res = self.testapp.get(
"/ns/test.example.com/import-comments", status=302
)
res = redirect_res.follow()
self.assertIn(b"You do not own that Namespace.", res.body)
# Clean up
self.dbsession.delete(other_user)
self.dbsession.flush()
self.tm.commit()
def test_import_page_displays_correctly(self):
"""Test that the import page displays correctly for namespace owners"""
res = self.testapp.get("/ns/test.example.com/import-comments", status=200)
self.assertIn(b"Import Comments", res.body)
self.assertIn(b"blog-to-json", res.body)
self.assertIn(b"How to Use", res.body)
self.assertIn(b"JSON File:", res.body)
self.assertIn(b"Automatic user creation", res.body)
def test_import_requires_file(self):
"""Test that import fails without a file upload"""
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
status=200,
)
self.assertIn(b"Please select a JSON file to upload.", res.body)
def test_import_invalid_json(self):
"""Test that import fails with invalid JSON"""
invalid_json_content = b"{ invalid json content"
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "test.json", invalid_json_content)],
status=200,
)
self.assertIn(b"Invalid JSON file", res.body)
def test_import_valid_json_with_comments(self):
"""Test successful import of comments from valid JSON"""
# Create a valid Disqus export JSON
disqus_data = {
"test-post": {
"link": "https://example.com/test-post",
"comments": [
{
"id": "1",
"author": "John Doe",
"email": "john@example.com",
"content": "This is a test comment",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
{
"id": "2",
"author": "Jane Smith",
"email": "jane@example.com",
"content": "This is a reply",
"timestamp": 1234567900,
"parent_id": "1",
"author_ip": "127.0.0.2",
},
],
}
}
json_content = json.dumps(disqus_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "disqus.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"1 threads", res.body)
self.assertIn(b"2 comments", res.body)
def test_import_with_user_surrogates(self):
"""Test import automatically creates surrogates for comments without email"""
disqus_data = {
"test-post": {
"link": "https://example.com/surrogate-test",
"comments": [
{
"id": "1",
"author": "Anonymous User",
"email": "",
"content": "Anonymous comment",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
],
}
}
json_content = json.dumps(disqus_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "disqus.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"1 comments", res.body)
def test_import_with_automatic_group_postfix(self):
"""Test import automatically adds group postfix to all new users"""
disqus_data = {
"test-post": {
"link": "https://example.com/group-test",
"comments": [
{
"id": "1",
"author": "Test User",
"email": "testgroup@example.com",
"content": "Test comment",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
],
}
}
json_content = json.dumps(disqus_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "disqus.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
# Verify user was created with automatic group postfix
user = get_user_by_email(self.dbsession, "testgroup@example.com")
self.assertIsNotNone(user)
# Should have postfix pattern in username (e.g., "Test-User-te")
self.assertIn("-", user.name)
def test_import_empty_comments(self):
"""Test import with threads that have no comments"""
disqus_data = {
"empty-post": {
"link": "https://example.com/empty-post",
"comments": [],
}
}
json_content = json.dumps(disqus_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "disqus.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"0 threads", res.body)
self.assertIn(b"0 comments", res.body)
def test_namespace_settings_links_to_import(self):
"""Test that namespace settings page has a link to import page"""
res = self.testapp.get("/ns/test.example.com/settings", status=200)
self.assertIn(b"Import Comments", res.body)
self.assertIn(b"import-comments", res.body)
def test_import_wordpress_format(self):
"""Test import from WordPress XML export (blog-to-json format)"""
# Mock WordPress export data converted by wordpress-xml-to-json
wordpress_data = {
"homegrown-python-bread-crumb-module": {
"name": "a-homegrown-python-bread-crumb-module",
"title": "A homegrown python bread crumb module",
"timestamp": 1293995686,
"link": "http://russell.ballestrini.net/a-homegrown-python-bread-crumb-module/",
"date": "2011-01-02 14:14:46",
"content": "<p>Some content here</p>",
"comments": [
{
"id": "wp-1",
"date": "2011-04-03 10:33:07",
"timestamp": 1301841187,
"content": "Hi, this was just what I needed",
"email": "kristian@example.com",
"author": "Kristian",
"author_ip": "192.168.1.5",
"parent_id": None,
},
{
"id": "wp-2",
"date": "2011-04-03 14:19:46",
"timestamp": 1301854786,
"content": "I'm interested in the modifications",
"email": "russell@example.com",
"author": "Russell Ballestrini",
"author_ip": "192.168.1.6",
"parent_id": None,
}
],
}
}
json_content = json.dumps(wordpress_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "wordpress.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"1 threads", res.body)
self.assertIn(b"2 comments", res.body)
# Verify users were created
kristian = get_user_by_email(self.dbsession, "kristian@example.com")
self.assertIsNotNone(kristian)
russell = get_user_by_email(self.dbsession, "russell@example.com")
self.assertIsNotNone(russell)
def test_import_graphcomment_format(self):
"""Test import from Graphcomment WordPress XML export (blog-to-json format)"""
# Mock Graphcomment export data converted by graphcomment-xml-to-json
graphcomment_data = {
"posts_jupyter-orgmode": {
"content": None,
"link": "https://abc.xyz/posts/jupyter-orgmode/",
"name": "posts_jupyter-orgmode",
"title": "Reflections on Jupyter",
"date": "2020-09-22 15:57:34",
"timestamp": 1600790254,
"id": "5f6a1eee2f57815d17188de2",
"metadata": {},
"comments": [
{
"id": "60750d2613ebd3704ec85f6f",
"content": "Thanks!!!!! a lot!!!",
"parent_id": None,
"author": "SamTux",
"date": "2021-04-13 03:16:54",
"timestamp": 1618283814,
"author_ip": "190.25.34.217",
"email": "samtux@example.com"
},
{
"id": "60750d2613ebd3704ec85f70",
"content": "You're welcome!",
"parent_id": "60750d2613ebd3704ec85f6f",
"author": "BlogAuthor",
"date": "2021-04-14 10:20:00",
"timestamp": 1618395600,
"author_ip": "192.168.1.1",
"email": "author@example.com"
}
],
}
}
json_content = json.dumps(graphcomment_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "graphcomment.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"1 threads", res.body)
self.assertIn(b"2 comments", res.body)
def test_import_with_deep_nesting(self):
"""Test import with deeply nested comment hierarchy"""
nested_data = {
"nested-discussion": {
"link": "https://example.com/nested-discussion",
"title": "Nested Discussion",
"timestamp": 1234567890,
"comments": [
{
"id": "1",
"author": "User1",
"email": "user1@example.com",
"content": "Top level comment",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
{
"id": "2",
"author": "User2",
"email": "user2@example.com",
"content": "Reply to comment 1",
"timestamp": 1234567900,
"parent_id": "1",
"author_ip": "127.0.0.2",
},
{
"id": "3",
"author": "User3",
"email": "user3@example.com",
"content": "Reply to comment 2",
"timestamp": 1234567910,
"parent_id": "2",
"author_ip": "127.0.0.3",
},
{
"id": "4",
"author": "User4",
"email": "user4@example.com",
"content": "Reply to comment 3",
"timestamp": 1234567920,
"parent_id": "3",
"author_ip": "127.0.0.4",
},
],
}
}
json_content = json.dumps(nested_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "nested.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"1 threads", res.body)
self.assertIn(b"4 comments", res.body)
def test_import_duplicate_prevention(self):
"""Test that re-importing the same data reuses existing users"""
test_data = {
"test-duplicate": {
"link": "https://example.com/test-duplicate",
"comments": [
{
"id": "dup-1",
"author": "DupUser",
"email": "dup@example.com",
"content": "First import",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
],
}
}
json_content = json.dumps(test_data).encode("utf-8")
# First import
res1 = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "dup.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res1.body)
self.assertIn(b"1 threads", res1.body)
self.assertIn(b"1 comments", res1.body)
# Second import - should succeed and reuse existing user
res2 = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "dup.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res2.body)
self.assertIn(b"1 threads", res2.body)
self.assertIn(b"1 comments", res2.body)
# Verify only one user was created (this check works across transactions)
from remarkbox.models import User
user_count = self.dbsession.query(User).filter(
User.email == "dup@example.com"
).count()
self.assertEqual(user_count, 1)
def test_import_with_locked_group_postfix(self):
"""Test that group postfix gets locked after first import"""
# Create a unique namespace for this test
from remarkbox.models import get_namespace_by_name
lock_test_ns = get_or_create_namespace(self.dbsession, "lock-test.example.com")
lock_test_ns.set_role_for_user(self.test_user, role="owner")
self.dbsession.add(lock_test_ns)
self.dbsession.flush()
self.tm.commit()
test_data = {
"test-lock": {
"link": "https://example.com/test-lock",
"comments": [
{
"id": "lock-1",
"author": "LockUser",
"email": "lock@example.com",
"content": "Test locking",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
],
}
}
json_content = json.dumps(test_data).encode("utf-8")
# First import - postfix should be set
res1 = self.testapp.post(
"/ns/lock-test.example.com/import-comments",
{
"csrf_token": self.csrf,
"group-prefix": "mygrp",
},
upload_files=[("json-file", "lock.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res1.body)
# Reload namespace to check postfix was locked
namespace = get_namespace_by_name(self.dbsession, "lock-test.example.com")
self.assertEqual(namespace.import_group_postfix, "mygrp")
# Second import should use locked postfix regardless of input
res2 = self.testapp.post(
"/ns/lock-test.example.com/import-comments",
{
"csrf_token": self.csrf,
"group-prefix": "different", # This should be ignored
},
upload_files=[("json-file", "lock2.json", json_content)],
status=200,
)
# Verify postfix didn't change
namespace = get_namespace_by_name(self.dbsession, "lock-test.example.com")
self.assertEqual(namespace.import_group_postfix, "mygrp")
def test_import_with_missing_email_creates_surrogates(self):
"""Test that comments without email create unique surrogates"""
test_data = {
"test-surrogates": {
"link": "https://example.com/test-surrogates",
"comments": [
{
"id": "s1",
"author": "Guest1",
"email": "",
"content": "First guest comment",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
{
"id": "s2",
"author": "Guest2",
"email": "",
"content": "Second guest comment",
"timestamp": 1234567900,
"parent_id": None,
"author_ip": "127.0.0.2",
},
{
"id": "s3",
"author": "Guest1", # Same name as first
"email": "",
"content": "Another comment from Guest1",
"timestamp": 1234567910,
"parent_id": None,
"author_ip": "127.0.0.3",
},
],
}
}
json_content = json.dumps(test_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "surrogates.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"3 comments", res.body)
# Verify surrogates were created
from remarkbox.models import UserSurrogate
namespace = self.get_test_namespace()
surrogates = self.dbsession.query(UserSurrogate).filter(
UserSurrogate.namespace_id == namespace.id
).all()
# Should have 2 unique surrogates (Guest1 and Guest2)
# Guest1 appearing twice should reuse the same surrogate
self.assertEqual(len(surrogates), 2)
# Verify surrogate names include group postfix
surrogate_names = [s.name for s in surrogates]
for name in surrogate_names:
self.assertIn("-", name) # Should have postfix
def test_import_with_missing_fields(self):
"""Test import handles missing optional fields gracefully"""
test_data = {
"test-missing-fields": {
"link": "https://example.com/test-missing",
"comments": [
{
"id": "m1",
"author": "MinimalUser",
"email": "minimal@example.com",
"content": "Minimal comment",
"timestamp": 1234567890,
# Missing parent_id
# Missing author_ip
},
],
}
}
json_content = json.dumps(test_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "missing.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"1 comments", res.body)
def test_import_multiple_threads(self):
"""Test importing multiple threads in one file"""
multi_thread_data = {
"thread-1": {
"link": "https://example.com/thread-1",
"comments": [
{
"id": "t1-c1",
"author": "User1",
"email": "user1@example.com",
"content": "Comment on thread 1",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
],
},
"thread-2": {
"link": "https://example.com/thread-2",
"comments": [
{
"id": "t2-c1",
"author": "User2",
"email": "user2@example.com",
"content": "Comment on thread 2",
"timestamp": 1234567900,
"parent_id": None,
"author_ip": "127.0.0.2",
},
],
},
"thread-3": {
"link": "https://example.com/thread-3",
"comments": [
{
"id": "t3-c1",
"author": "User3",
"email": "user3@example.com",
"content": "Comment on thread 3",
"timestamp": 1234567910,
"parent_id": None,
"author_ip": "127.0.0.3",
},
],
},
}
json_content = json.dumps(multi_thread_data).encode("utf-8")
res = self.testapp.post(
"/ns/test.example.com/import-comments",
{
"csrf_token": self.csrf,
},
upload_files=[("json-file", "multi.json", json_content)],
status=200,
)
self.assertIn(b"Successfully imported", res.body)
self.assertIn(b"3 threads", res.body)
self.assertIn(b"3 comments", res.body)
def test_import_with_invalid_group_postfix(self):
"""Test that invalid group postfix shows error"""
# Create a unique namespace for this test
invalid_test_ns = get_or_create_namespace(self.dbsession, "invalid-test.example.com")
invalid_test_ns.set_role_for_user(self.test_user, role="owner")
self.dbsession.add(invalid_test_ns)
self.dbsession.flush()
self.tm.commit()
test_data = {
"test": {
"link": "https://example.com/test",
"comments": [
{
"id": "1",
"author": "User",
"email": "user@example.com",
"content": "Test",
"timestamp": 1234567890,
"parent_id": None,
"author_ip": "127.0.0.1",
},
],
}
}
json_content = json.dumps(test_data).encode("utf-8")
# Try with too short postfix
res = self.testapp.post(
"/ns/invalid-test.example.com/import-comments",
{
"csrf_token": self.csrf,
"group-prefix": "x", # Only 1 char, minimum is 2
},
upload_files=[("json-file", "invalid.json", json_content)],
status=200,
)
self.assertIn(b"Group postfix is required and must be at least 2 alphanumeric characters", res.body)
class ImportCommentsUnitTests(unittest.TestCase):
"""Unit tests for import_comments view functions"""
def test_generate_password(self):
"""Test password generation utility"""
from remarkbox.views.authenticated.import_comments import generate_password
password = generate_password(32)
self.assertEqual(len(password), 32)
self.assertTrue(all(c.isalnum() for c in password))
def test_generate_password_custom_size(self):
"""Test password generation with custom size"""
from remarkbox.views.authenticated.import_comments import generate_password
password = generate_password(16)
self.assertEqual(len(password), 16)
def test_generate_group_prefix_from_namespace(self):
"""Test group prefix generation from namespace domain"""
from remarkbox.views.authenticated.import_comments import generate_group_prefix_from_namespace
# Test multi-part domain
self.assertEqual(generate_group_prefix_from_namespace("russell.ballestrini.net"), "rb")
# Test with www prefix
self.assertEqual(generate_group_prefix_from_namespace("www.example.com"), "exampl")
# Test single-part domain
self.assertEqual(generate_group_prefix_from_namespace("example.com"), "exampl")
# Test my.remarkbox.com style
prefix = generate_group_prefix_from_namespace("my.remarkbox.com")
self.assertTrue(len(prefix) <= 6)
self.assertTrue(prefix.isalnum())

276
remarkbox/views/authenticated/import_comments.py Normal file
View file

@ -0,0 +1,276 @@
from pyramid.view import view_config
from pyramid.httpexceptions import HTTPFound
import json
import tempfile
import os
from datetime import datetime
from remarkbox.models import (
User,
UserSurrogate,
generate_user_name,
get_user_by_email,
get_user_surrogate_by_name,
get_or_create_node_by_uri,
is_user_name_valid,
is_user_name_available,
)
from remarkbox.views import get_referer_or_home, user_required
try:
unicode("")
except:
from six import u as unicode
def generate_password(size=32):
"""Return a system generated password"""
from random import choice
letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
digits = "0123456789"
pool = letters + digits
return "".join([choice(pool) for i in range(size)])
def generate_group_prefix_from_namespace(namespace_name):
"""
Generate a short, safe group prefix from namespace domain.
Takes first letters of domain parts, max 6 chars.
Example: russell.ballestrini.net -> rb, my.remarkbox.com -> mrc
"""
# Remove common TLDs and split on dots
domain = namespace_name.lower()
# Remove www prefix if present
if domain.startswith('www.'):
domain = domain[4:]
# Split on dots and filter out common TLDs
parts = domain.split('.')
# Remove TLDs like com, org, net, io, etc
tlds = {'com', 'org', 'net', 'io', 'co', 'edu', 'gov', 'mil', 'int'}
parts = [p for p in parts if p not in tlds]
if not parts:
# Use first 6 chars of full domain if no valid parts
prefix = ''.join(c for c in namespace_name if c.isalnum())[:6]
if len(prefix) < 2:
# Absolutely no valid characters, use hash of namespace
import hashlib
prefix = hashlib.md5(namespace_name.encode()).hexdigest()[:6]
return prefix
# Strategy 1: Take first letter of each part (e.g., russell.ballestrini -> rb)
prefix = ''.join(p[0] for p in parts if p)
# Strategy 2: If too short or just one part, take first few chars
if len(prefix) < 2 or len(parts) == 1:
prefix = parts[0][:6]
# Ensure max 6 chars and only alphanumeric
prefix = ''.join(c for c in prefix if c.isalnum())[:6]
# Final safety check - ensure at least 2 chars
if len(prefix) < 2:
# Use hash of namespace as last resort
import hashlib
prefix = hashlib.md5(namespace_name.encode()).hexdigest()[:6]
return prefix
@view_config(route_name="basic-namespace-import-comments", renderer="import-comments.j2")
@view_config(route_name="embed-namespace-import-comments", renderer="import-comments.j2")
@user_required()
def import_comments(request):
"""
Import comments and threads from JSON dumps created by the blog-to-json tool.
Supports WordPress, Disqus, and pre-formatted JSON exports.
This view handles file upload and processes the JSON data.
"""
if not request.user in request.namespace.owners:
request.session.flash(("You do not own that Namespace.", "error"))
return HTTPFound(get_referer_or_home(request))
# Check if namespace already has a locked group postfix
if request.namespace.import_group_postfix:
# Use the locked postfix
default_postfix = request.namespace.import_group_postfix
postfix_locked = True
else:
# Generate default group postfix from namespace
default_postfix = generate_group_prefix_from_namespace(request.namespace.name)
postfix_locked = False
if request.method == "POST":
# Get the group postfix, allow user to override if not locked
if postfix_locked:
group = request.namespace.import_group_postfix
else:
group_postfix = request.params.get("group-prefix", "").strip()
if not group_postfix:
group_postfix = default_postfix
# Sanitize: max 6 chars, alphanumeric only
group = ''.join(c for c in group_postfix if c.isalnum())[:6]
if not group or len(group) < 2:
request.session.flash(("Group postfix is required and must be at least 2 alphanumeric characters.", "error"))
return {
"the_title": "Import Comments",
"default_prefix": default_postfix,
"postfix_locked": postfix_locked,
}
# Lock the group postfix in the namespace on first use
request.namespace.import_group_postfix = group
request.dbsession.add(request.namespace)
request.dbsession.flush()
# Get the uploaded file
upload_file = request.params.get("json-file", None)
if upload_file is None or upload_file == b'':
request.session.flash(("Please select a JSON file to upload.", "error"))
return {"the_title": "Import Comments"}
try:
# Read the uploaded file
json_content = upload_file.file.read()
# Parse JSON
try:
pages = json.loads(json_content)
except json.JSONDecodeError as e:
request.session.flash((f"Invalid JSON file: {str(e)}", "error"))
return {"the_title": "Import Comments"}
# Process the import
users = {}
user_surrogates = {}
nodes = {}
imported_threads = 0
imported_comments = 0
for slug, page_data in pages.items():
comments = page_data.get("comments", [])
if len(comments) == 0:
continue
uri = page_data.get("link", "")
if not uri:
continue
# Create the root node for this thread
root = get_or_create_node_by_uri(request.dbsession, uri)
root.namespace = request.namespace
request.dbsession.add(root)
request.dbsession.flush()
imported_threads += 1
for comment in comments:
email = comment.get("email", "")
user = None
user_surrogate = None
if not email:
# Always create surrogates for comments without email
# Use group postfix to make them identifiable
author_name = comment.get("author", "Anonymous")
# Create full name with group postfix (e.g., "Anonymous-rb-20231122-143045")
surrogate_name_with_group = f"{author_name}-{group}"
# Check in-memory cache first
if surrogate_name_with_group not in user_surrogates:
# Check if this surrogate already exists in the database
existing_surrogate = get_user_surrogate_by_name(
request.dbsession,
surrogate_name_with_group,
root.namespace
)
if existing_surrogate:
user_surrogates[surrogate_name_with_group] = existing_surrogate
else:
# Create new surrogate
user_surrogates[surrogate_name_with_group] = UserSurrogate(
surrogate_name_with_group,
root.namespace,
)
request.dbsession.add(user_surrogates[surrogate_name_with_group])
request.dbsession.flush()
user_surrogate = user_surrogates[surrogate_name_with_group]
elif email in users:
user = users[email]
else:
user = get_user_by_email(request.dbsession, email)
if user:
users[email] = user
else:
desired_name = (
comment.get("author", "Anonymous").replace(" ", "-").replace("_", "-")
)
# Always append group name to identify imported users
desired_name = desired_name + "-" + group
if not is_user_name_valid(
desired_name
) or not is_user_name_available(
request.dbsession, desired_name
):
desired_name = generate_user_name(
request.dbsession, desired_name
)
user = User(email)
user.name = unicode(desired_name)
users[email] = user
request.dbsession.add(user)
request.dbsession.flush()
# Create the comment node
node = root.new_child()
node.set_data(unicode(comment.get("content", "")))
node.created = int(comment.get("timestamp", 0)) * 1000
node.changed = int(comment.get("timestamp", 0)) * 1000
if user is not None:
node.user = user
elif user_surrogate is not None:
node.user_surrogate = user_surrogate
node.verified = True
node.ip_address = comment.get("author_ip", None)
nodes[comment.get("id", "")] = node
request.dbsession.add(node)
request.dbsession.flush()
imported_comments += 1
# Set up parent-child relationships
for comment in comments:
parent_id = comment.get("parent_id")
comment_id = comment.get("id", "")
if parent_id and nodes.get(comment_id) and nodes.get(parent_id):
nodes[comment_id].parent_id = nodes[parent_id].id
request.dbsession.add(nodes[comment_id])
request.dbsession.flush()
request.session.flash((
f"Successfully imported {imported_threads} threads and {imported_comments} comments.",
"success"
))
except Exception as e:
request.session.flash((f"Error during import: {str(e)}", "error"))
return {
"the_title": "Import Comments",
"default_prefix": default_postfix,
"postfix_locked": postfix_locked,
}