Speed up daily reindex: skip empty JSON rewrites and upsert only changed posts.
Keep channel mtime stable when merge finds no new posts, and rebuild FTS only for new/changed content so large daily channels stay cheap. Drop --reindex-all from renew.sh for incremental runs. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -624,6 +624,10 @@ def export_channel(d: Driver, channel: str, user_id_to_name: Dict[str, str], out
|
|||||||
merged_posts.append(post)
|
merged_posts.append(post)
|
||||||
seen_ids.add(post["id"])
|
seen_ids.add(post["id"])
|
||||||
new_count += 1
|
new_count += 1
|
||||||
|
if new_count == 0:
|
||||||
|
# Keep mtime stable so mm_index can skip this channel file.
|
||||||
|
print(f" no new posts, leaving '{output_filepath}' unchanged")
|
||||||
|
return
|
||||||
merged_posts.sort(key=lambda p: p["created"])
|
merged_posts.sort(key=lambda p: p["created"])
|
||||||
for i, post in enumerate(merged_posts):
|
for i, post in enumerate(merged_posts):
|
||||||
post["idx"] = i
|
post["idx"] = i
|
||||||
|
|||||||
+168
-57
@@ -143,10 +143,102 @@ def load_posts(path: str) -> Optional[Tuple[Dict[str, Any], List[Dict[str, Any]]
|
|||||||
return channel_obj, posts
|
return channel_obj, posts
|
||||||
|
|
||||||
|
|
||||||
def upsert_file(conn: sqlite3.Connection, path: str) -> Tuple[int, int]:
|
def _delete_post_search_rows(conn: sqlite3.Connection, post_id: str) -> None:
|
||||||
|
conn.execute("DELETE FROM attachments_fts WHERE post_id = ?", (post_id,))
|
||||||
|
conn.execute("DELETE FROM attachments WHERE post_id = ?", (post_id,))
|
||||||
|
conn.execute("DELETE FROM posts_fts WHERE post_id = ?", (post_id,))
|
||||||
|
|
||||||
|
|
||||||
|
def _delete_post(conn: sqlite3.Connection, post_id: str) -> None:
|
||||||
|
_delete_post_search_rows(conn, post_id)
|
||||||
|
conn.execute("DELETE FROM posts WHERE post_id = ?", (post_id,))
|
||||||
|
|
||||||
|
|
||||||
|
def _index_attachments(
|
||||||
|
conn: sqlite3.Connection,
|
||||||
|
post_id: str,
|
||||||
|
files: Any,
|
||||||
|
idx: Optional[int],
|
||||||
|
base_dir: str,
|
||||||
|
max_text_size: int,
|
||||||
|
) -> None:
|
||||||
|
if not isinstance(files, list) or not files:
|
||||||
|
return
|
||||||
|
for filename in files:
|
||||||
|
if not isinstance(filename, str) or not filename:
|
||||||
|
continue
|
||||||
|
disk_name = f"{idx:03d}_{filename}" if isinstance(idx, int) else filename
|
||||||
|
disk_path = os.path.join(base_dir, disk_name)
|
||||||
|
is_text = 0
|
||||||
|
content: Optional[str] = None
|
||||||
|
try:
|
||||||
|
st = os.stat(disk_path)
|
||||||
|
if st.st_size <= max_text_size:
|
||||||
|
with open(disk_path, "r", encoding="utf-8") as f:
|
||||||
|
content = f.read()
|
||||||
|
is_text = 1
|
||||||
|
except (OSError, UnicodeDecodeError):
|
||||||
|
is_text = 0
|
||||||
|
content = None
|
||||||
|
|
||||||
|
cur = conn.execute(
|
||||||
|
"""
|
||||||
|
INSERT INTO attachments (post_id, filename, filepath, is_text)
|
||||||
|
VALUES (?, ?, ?, ?)
|
||||||
|
""",
|
||||||
|
(post_id, filename, disk_path, is_text),
|
||||||
|
)
|
||||||
|
attachment_id = cur.lastrowid
|
||||||
|
conn.execute(
|
||||||
|
"""
|
||||||
|
INSERT INTO attachments_fts (attachment_id, post_id, filename, content)
|
||||||
|
VALUES (?, ?, ?, ?)
|
||||||
|
""",
|
||||||
|
(attachment_id, post_id, filename, content if is_text else None),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _content_changed(prev: sqlite3.Row, message: str, username: str, created: str, files_json: Optional[str]) -> bool:
|
||||||
|
return (
|
||||||
|
prev["message"] != message
|
||||||
|
or prev["username"] != username
|
||||||
|
or prev["created"] != created
|
||||||
|
or prev["files_json"] != files_json
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _meta_changed(
|
||||||
|
prev: sqlite3.Row,
|
||||||
|
idx: Optional[int],
|
||||||
|
channel_id: str,
|
||||||
|
channel_name: str,
|
||||||
|
channel_display_name: str,
|
||||||
|
team: str,
|
||||||
|
path: str,
|
||||||
|
exported_at: Optional[str],
|
||||||
|
) -> bool:
|
||||||
|
return (
|
||||||
|
prev["idx"] != idx
|
||||||
|
or prev["channel_id"] != channel_id
|
||||||
|
or prev["channel_name"] != channel_name
|
||||||
|
or prev["channel_display_name"] != channel_display_name
|
||||||
|
or prev["team"] != team
|
||||||
|
or prev["source_path"] != path
|
||||||
|
or prev["exported_at"] != exported_at
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def upsert_file(conn: sqlite3.Connection, path: str, *, force: bool = False) -> Tuple[int, int, int]:
|
||||||
|
"""Index posts from one export JSON.
|
||||||
|
|
||||||
|
Returns (upserted_or_deleted, invalid_skipped, unchanged).
|
||||||
|
When force=False, only new/changed posts rebuild FTS/attachments; idx-only
|
||||||
|
renumbers get a cheap posts-row UPDATE. force=True rebuilds every post
|
||||||
|
(--reindex-all).
|
||||||
|
"""
|
||||||
loaded = load_posts(path)
|
loaded = load_posts(path)
|
||||||
if loaded is None:
|
if loaded is None:
|
||||||
return 0, 0
|
return 0, 0, 0
|
||||||
|
|
||||||
channel_obj, posts = loaded
|
channel_obj, posts = loaded
|
||||||
channel_id = str(channel_obj.get("id") or "")
|
channel_id = str(channel_obj.get("id") or "")
|
||||||
@@ -155,27 +247,28 @@ def upsert_file(conn: sqlite3.Connection, path: str) -> Tuple[int, int]:
|
|||||||
team = str(channel_obj.get("team") or "")
|
team = str(channel_obj.get("team") or "")
|
||||||
exported_at = parse_exported_at(channel_obj)
|
exported_at = parse_exported_at(channel_obj)
|
||||||
|
|
||||||
# Remove any existing attachments for posts coming from this source file
|
existing = {
|
||||||
conn.execute(
|
row["post_id"]: row
|
||||||
"""
|
for row in conn.execute(
|
||||||
DELETE FROM attachments
|
"""
|
||||||
WHERE post_id IN (SELECT post_id FROM posts WHERE source_path = ?)
|
SELECT post_id, idx, created, username, message, files_json,
|
||||||
""",
|
channel_id, channel_name, channel_display_name, team,
|
||||||
(path,),
|
source_path, exported_at
|
||||||
)
|
FROM posts
|
||||||
conn.execute(
|
WHERE source_path = ?
|
||||||
"""
|
""",
|
||||||
DELETE FROM attachments_fts
|
(path,),
|
||||||
WHERE post_id IN (SELECT post_id FROM posts WHERE source_path = ?)
|
)
|
||||||
""",
|
}
|
||||||
(path,),
|
|
||||||
)
|
|
||||||
|
|
||||||
base_dir = os.path.dirname(path)
|
base_dir = os.path.dirname(path)
|
||||||
max_text_size = 2 * 1024 * 1024 # 2 MiB
|
max_text_size = 2 * 1024 * 1024 # 2 MiB
|
||||||
|
|
||||||
inserted_or_updated = 0
|
inserted_or_updated = 0
|
||||||
|
unchanged = 0
|
||||||
skipped = 0
|
skipped = 0
|
||||||
|
seen_ids = set()
|
||||||
|
|
||||||
for p in posts:
|
for p in posts:
|
||||||
post_id = p.get("id")
|
post_id = p.get("id")
|
||||||
created_raw = p.get("created")
|
created_raw = p.get("created")
|
||||||
@@ -198,6 +291,49 @@ def upsert_file(conn: sqlite3.Connection, path: str) -> Tuple[int, int]:
|
|||||||
files = p.get("files")
|
files = p.get("files")
|
||||||
files_json = json.dumps(files, ensure_ascii=False) if isinstance(files, list) else None
|
files_json = json.dumps(files, ensure_ascii=False) if isinstance(files, list) else None
|
||||||
idx = p.get("idx") if isinstance(p.get("idx"), int) else None
|
idx = p.get("idx") if isinstance(p.get("idx"), int) else None
|
||||||
|
seen_ids.add(post_id)
|
||||||
|
prev = existing.get(post_id)
|
||||||
|
|
||||||
|
if (
|
||||||
|
not force
|
||||||
|
and prev is not None
|
||||||
|
and not _content_changed(prev, message, username, created, files_json)
|
||||||
|
):
|
||||||
|
if _meta_changed(
|
||||||
|
prev,
|
||||||
|
idx,
|
||||||
|
channel_id,
|
||||||
|
channel_name,
|
||||||
|
channel_display_name,
|
||||||
|
team,
|
||||||
|
path,
|
||||||
|
exported_at,
|
||||||
|
):
|
||||||
|
conn.execute(
|
||||||
|
"""
|
||||||
|
UPDATE posts SET
|
||||||
|
idx = ?,
|
||||||
|
channel_id = ?,
|
||||||
|
channel_name = ?,
|
||||||
|
channel_display_name = ?,
|
||||||
|
team = ?,
|
||||||
|
source_path = ?,
|
||||||
|
exported_at = ?
|
||||||
|
WHERE post_id = ?
|
||||||
|
""",
|
||||||
|
(
|
||||||
|
idx,
|
||||||
|
channel_id,
|
||||||
|
channel_name,
|
||||||
|
channel_display_name,
|
||||||
|
team,
|
||||||
|
path,
|
||||||
|
exported_at,
|
||||||
|
post_id,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
unchanged += 1
|
||||||
|
continue
|
||||||
|
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"""
|
"""
|
||||||
@@ -234,45 +370,16 @@ def upsert_file(conn: sqlite3.Connection, path: str) -> Tuple[int, int]:
|
|||||||
exported_at,
|
exported_at,
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
conn.execute("DELETE FROM posts_fts WHERE post_id = ?", (post_id,))
|
_delete_post_search_rows(conn, post_id)
|
||||||
conn.execute("INSERT INTO posts_fts(post_id, message) VALUES (?, ?)", (post_id, message))
|
conn.execute("INSERT INTO posts_fts(post_id, message) VALUES (?, ?)", (post_id, message))
|
||||||
|
_index_attachments(conn, post_id, files, idx, base_dir, max_text_size)
|
||||||
# Index attachments (file names and, for text files, contents)
|
|
||||||
if isinstance(files, list) and files:
|
|
||||||
for filename in files:
|
|
||||||
if not isinstance(filename, str) or not filename:
|
|
||||||
continue
|
|
||||||
disk_name = f"{idx:03d}_{filename}" if isinstance(idx, int) else filename
|
|
||||||
disk_path = os.path.join(base_dir, disk_name)
|
|
||||||
is_text = 0
|
|
||||||
content: Optional[str] = None
|
|
||||||
try:
|
|
||||||
st = os.stat(disk_path)
|
|
||||||
if st.st_size <= max_text_size:
|
|
||||||
with open(disk_path, "r", encoding="utf-8") as f:
|
|
||||||
content = f.read()
|
|
||||||
is_text = 1
|
|
||||||
except (OSError, UnicodeDecodeError):
|
|
||||||
is_text = 0
|
|
||||||
content = None
|
|
||||||
|
|
||||||
cur = conn.execute(
|
|
||||||
"""
|
|
||||||
INSERT INTO attachments (post_id, filename, filepath, is_text)
|
|
||||||
VALUES (?, ?, ?, ?)
|
|
||||||
""",
|
|
||||||
(post_id, filename, disk_path, is_text),
|
|
||||||
)
|
|
||||||
attachment_id = cur.lastrowid
|
|
||||||
conn.execute(
|
|
||||||
"""
|
|
||||||
INSERT INTO attachments_fts (attachment_id, post_id, filename, content)
|
|
||||||
VALUES (?, ?, ?, ?)
|
|
||||||
""",
|
|
||||||
(attachment_id, post_id, filename, content if is_text else None),
|
|
||||||
)
|
|
||||||
inserted_or_updated += 1
|
inserted_or_updated += 1
|
||||||
|
|
||||||
|
for post_id in existing:
|
||||||
|
if post_id not in seen_ids:
|
||||||
|
_delete_post(conn, post_id)
|
||||||
|
inserted_or_updated += 1
|
||||||
|
|
||||||
st = os.stat(path)
|
st = os.stat(path)
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"""
|
"""
|
||||||
@@ -285,7 +392,7 @@ def upsert_file(conn: sqlite3.Connection, path: str) -> Tuple[int, int]:
|
|||||||
""",
|
""",
|
||||||
(path, st.st_mtime_ns, st.st_size, dt.datetime.now(dt.timezone.utc).isoformat()),
|
(path, st.st_mtime_ns, st.st_size, dt.datetime.now(dt.timezone.utc).isoformat()),
|
||||||
)
|
)
|
||||||
return inserted_or_updated, skipped
|
return inserted_or_updated, skipped, unchanged
|
||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
def main() -> int:
|
||||||
@@ -302,6 +409,7 @@ def main() -> int:
|
|||||||
skipped_files = 0
|
skipped_files = 0
|
||||||
rows_upserted = 0
|
rows_upserted = 0
|
||||||
rows_skipped = 0
|
rows_skipped = 0
|
||||||
|
rows_unchanged = 0
|
||||||
|
|
||||||
with conn:
|
with conn:
|
||||||
for path in files:
|
for path in files:
|
||||||
@@ -310,9 +418,10 @@ def main() -> int:
|
|||||||
skipped_files += 1
|
skipped_files += 1
|
||||||
continue
|
continue
|
||||||
processed_files += 1
|
processed_files += 1
|
||||||
upserted, skipped = upsert_file(conn, path)
|
upserted, skipped, unchanged = upsert_file(conn, path, force=args.reindex_all)
|
||||||
rows_upserted += upserted
|
rows_upserted += upserted
|
||||||
rows_skipped += skipped
|
rows_skipped += skipped
|
||||||
|
rows_unchanged += unchanged
|
||||||
|
|
||||||
users_json = mm_users.find_users_json(args.root)
|
users_json = mm_users.find_users_json(args.root)
|
||||||
if users_json:
|
if users_json:
|
||||||
@@ -324,11 +433,13 @@ def main() -> int:
|
|||||||
|
|
||||||
print(f"DB: {args.db}")
|
print(f"DB: {args.db}")
|
||||||
print(f"Files scanned: {len(files)} (processed={processed_files}, unchanged_skipped={skipped_files})")
|
print(f"Files scanned: {len(files)} (processed={processed_files}, unchanged_skipped={skipped_files})")
|
||||||
print(f"Rows upserted: {rows_upserted}, invalid_posts_skipped: {rows_skipped}")
|
print(
|
||||||
|
f"Rows upserted: {rows_upserted}, unchanged_posts: {rows_unchanged}, "
|
||||||
|
f"invalid_posts_skipped: {rows_skipped}"
|
||||||
|
)
|
||||||
print(f"Totals: posts={total_posts}, sources={total_sources}")
|
print(f"Totals: posts={total_posts}, sources={total_sources}")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
raise SystemExit(main())
|
raise SystemExit(main())
|
||||||
|
|
||||||
|
|||||||
@@ -1,2 +1,4 @@
|
|||||||
cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mattermost-dl.py --team esr-daily --channel all
|
cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mattermost-dl.py --team esr-daily --channel all
|
||||||
cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mm_index.py --reindex-all
|
cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mattermost-dl.py --team esr-dev-and-qa --channel QoSd
|
||||||
|
cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mattermost-dl.py --team esr-dev-and-qa --channel VSAT
|
||||||
|
cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mm_index.py
|
||||||
|
|||||||
Reference in New Issue
Block a user