# FILE: _St/Shared/storage/sort_storage.py | ROLE: sort 공통 SQLite 스키마·조회·저장
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sqlite3
import urllib.parse
import uuid
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Mapping
SCHEMA_VERSION = 7
SCHEMA_FILES = (
Path(__file__).with_name("schema_sqlite_v1.sql"),
Path(__file__).with_name("schema_sqlite_v2.sql"),
Path(__file__).with_name("schema_sqlite_v3.sql"),
Path(__file__).with_name("schema_sqlite_v4.sql"),
Path(__file__).with_name("schema_sqlite_v5.sql"),
Path(__file__).with_name("schema_sqlite_v6.sql"),
Path(__file__).with_name("schema_sqlite_v7.sql"),
)
DEFAULT_DB_FILENAME = "sort_local.sqlite3"
DEFAULT_DB_RELATIVE = Path("Shared") / "data" / DEFAULT_DB_FILENAME
DATABASE_SELECTION_RELATIVE = Path("Shared") / "data" / "sort_database_selection.json"
_DATABASE_FILENAME_RE = re.compile(r"^[0-9A-Za-z가-힣._-]+$")
DEFAULT_STORAGE_CODE = "STDOWN_INSTAGRAM"
SOURCE_MODES = frozenset(
{
"MANUAL_SINGLE",
"BULK_ALL",
"BULK_FILTERED",
"TOP_SORTED",
"TRANSCRIPT_REQUEST",
"AUTO_SEEN_JPG",
"CACHE8701_CDP",
}
)
IMAGE_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp", ".avif", ".gif")
VIDEO_EXTENSIONS = (".mp4",)
def derive_stdown_root(st_root: str | Path) -> Path:
"""Return the fixed sibling _StDown directory for the supplied _St root."""
root = Path(st_root).expanduser().resolve()
if root.name != "_St":
raise ValueError(f"st_root must point to the _St directory: {root}")
return root.parent / "_StDown"
def safe_media_component(value: object, fallback: str = "unknown") -> str:
text = re.sub(r'[^0-9A-Za-z._-]+', "_", str(value or "").strip()).strip("._")
return text[:120] or fallback
def normalize_media_extension(value: object) -> str:
extension = str(value or "").strip().lower()
if extension and not extension.startswith("."):
extension = "." + extension
allowed = set(IMAGE_EXTENSIONS) | set(VIDEO_EXTENSIONS)
if extension not in allowed:
raise ValueError(f"unsupported media extension: {value!r}")
return extension
def build_instagram_media_path(
st_root: str | Path,
account_name: object,
shortcode: object,
extension: object,
) -> Path:
account = safe_media_component(account_name, "instagram")
code = safe_media_component(shortcode, "post")
ext = normalize_media_extension(extension)
return derive_stdown_root(st_root) / "instagram" / account / f"{code}{ext}"
def ensure_instagram_account_index(st_root: str | Path, account_name: object) -> Path:
"""Create one minimal account index that opens the account view in Analyze8780."""
account = safe_media_component(account_name, "instagram")
instagram_root = derive_stdown_root(st_root) / "instagram"
folder = instagram_root / account
folder.mkdir(parents=True, exist_ok=True)
# v303 [Cache Image 회귀 금지]: SQLite 이전의 해시·계정 JSON·정적 보고서 산출물만 정리한다.
retired = [
instagram_root / f"{account}.json",
instagram_root / f"{account}.html",
folder / f"{account}.json",
folder / "echarts.min.js",
]
retired.extend(folder.glob("*.md5"))
for legacy_path in retired:
try:
if legacy_path.is_file():
legacy_path.unlink()
except OSError:
pass
url = f"http://127.0.0.1:8780/?account={urllib.parse.quote(account)}"
html = (
f"\n"
"\n"
""
f"{account} · Analyze8780"
f""
""
f""
f"{account} Analyze8780 열기"
"\n"
)
path = folder / "index.htm"
if not path.is_file() or path.read_text(encoding="utf-8", errors="replace") != html:
temp = path.with_suffix(".htm.tmp")
temp.write_text(html, encoding="utf-8")
temp.replace(path)
return path
def find_instagram_media_file(
st_root: str | Path,
account_name: object,
shortcode: object,
*,
asset_kind: str = "IMAGE",
) -> Path | None:
account = safe_media_component(account_name, "instagram")
code = safe_media_component(shortcode, "post")
folder = derive_stdown_root(st_root) / "instagram" / account
if not folder.is_dir():
return None
kind = str(asset_kind or "IMAGE").strip().upper()
if kind in {"VIDEO", "MP4"}:
names = [f"{code}{ext}" for ext in VIDEO_EXTENSIONS]
else:
names = [f"{code}{ext}" for ext in IMAGE_EXTENSIONS]
for name in names:
candidate = folder / name
if candidate.is_file() and candidate.stat().st_size > 0:
return candidate
return None
def database_directory(st_root: str | Path) -> Path:
return Path(st_root).expanduser().resolve() / "Shared" / "data"
def database_selection_path(st_root: str | Path) -> Path:
return Path(st_root).expanduser().resolve() / DATABASE_SELECTION_RELATIVE
def normalize_database_filename(value: object) -> str:
raw = str(value or "").strip()
if not raw:
raise ValueError("database filename required")
if "/" in raw or "\\" in raw or Path(raw).name != raw:
raise ValueError("database filename must not contain a path")
name = raw
if not _DATABASE_FILENAME_RE.fullmatch(name):
raise ValueError("database filename contains unsupported characters")
if not name.lower().endswith(".sqlite3"):
name += ".sqlite3"
if name in {".", ".."} or name.startswith("."):
raise ValueError("database filename is invalid")
return name
def read_database_selection(st_root: str | Path) -> str:
path = database_selection_path(st_root)
try:
payload = json.loads(path.read_text(encoding="utf-8"))
return normalize_database_filename(payload.get("filename") or DEFAULT_DB_FILENAME)
except FileNotFoundError:
return DEFAULT_DB_FILENAME
except Exception:
return DEFAULT_DB_FILENAME
def write_database_selection(st_root: str | Path, filename: object) -> Path:
root = Path(st_root).expanduser().resolve()
name = normalize_database_filename(filename)
path = database_selection_path(root)
path.parent.mkdir(parents=True, exist_ok=True)
temp = path.with_suffix(path.suffix + ".tmp")
temp.write_text(json.dumps({"filename": name}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
temp.replace(path)
return database_directory(root) / name
def default_database_path(st_root: str | Path) -> Path:
root = Path(st_root).expanduser().resolve()
return database_directory(root) / read_database_selection(root)
def list_database_files(st_root: str | Path) -> dict[str, Any]:
root = Path(st_root).expanduser().resolve()
directory = database_directory(root)
directory.mkdir(parents=True, exist_ok=True)
active = read_database_selection(root)
files = sorted((path.name for path in directory.glob("*.sqlite3") if path.is_file()), key=str.casefold)
if DEFAULT_DB_FILENAME not in files and not files:
files = [DEFAULT_DB_FILENAME]
return {"active_filename": active, "active_path": str((directory / active).resolve()), "files": files}
def open_database_file(st_root: str | Path, filename: object) -> dict[str, Any]:
root = Path(st_root).expanduser().resolve()
name = normalize_database_filename(filename)
target = database_directory(root) / name
if not target.is_file():
raise ValueError(f"database file not found: {name}")
initialize_database(target, root)
write_database_selection(root, name)
return {**list_database_files(root), "db_path": str(target.resolve()), "filename": name}
def create_database_file(st_root: str | Path, filename: object) -> dict[str, Any]:
root = Path(st_root).expanduser().resolve()
name = normalize_database_filename(filename)
target = database_directory(root) / name
if target.exists():
raise ValueError(f"database file already exists: {name}")
initialize_database(target, root)
write_database_selection(root, name)
return {**list_database_files(root), "db_path": str(target.resolve()), "filename": name}
def save_database_as(st_root: str | Path, source_db_path: str | Path, filename: object) -> dict[str, Any]:
root = Path(st_root).expanduser().resolve()
name = normalize_database_filename(filename)
source = Path(source_db_path).expanduser().resolve()
target = database_directory(root) / name
if target.exists():
raise ValueError(f"database file already exists: {name}")
initialize_database(source, root)
target.parent.mkdir(parents=True, exist_ok=True)
src = connect_database(source)
dst = sqlite3.connect(target, timeout=10.0)
try:
src.backup(dst)
dst.commit()
except Exception:
dst.close()
src.close()
try:
target.unlink(missing_ok=True)
except Exception:
pass
raise
else:
dst.close()
src.close()
initialize_database(target, root)
write_database_selection(root, name)
return {**list_database_files(root), "db_path": str(target.resolve()), "filename": name}
def connect_database(db_path: str | Path) -> sqlite3.Connection:
path = Path(db_path).expanduser().resolve()
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(path, timeout=10.0)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA busy_timeout = 10000")
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")
return conn
def _table_columns(conn: sqlite3.Connection, table_name: str) -> set[str]:
return {str(row["name"]) for row in conn.execute(f"PRAGMA table_info({table_name})")}
def _ensure_column(conn: sqlite3.Connection, table_name: str, column_name: str, definition: str) -> None:
if column_name not in _table_columns(conn, table_name):
conn.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {definition}")
def _apply_schema(conn: sqlite3.Connection) -> None:
for schema_file in SCHEMA_FILES:
conn.executescript(schema_file.read_text(encoding="utf-8"))
_ensure_column(conn, "social_accounts", "deleted_at", "TEXT")
_ensure_column(conn, "posts", "deleted_at", "TEXT")
_ensure_column(conn, "media_assets", "folder_name", "TEXT NOT NULL DEFAULT ''")
_ensure_column(conn, "media_saved_events", "folder_name", "TEXT NOT NULL DEFAULT ''")
for table_name in ("post_stat_latest", "post_stat_history", "post_stat_10m"):
_ensure_column(conn, table_name, "sort_by", "TEXT NOT NULL DEFAULT ''")
_ensure_column(conn, table_name, "sort_value", "INTEGER NOT NULL DEFAULT 0")
_ensure_column(conn, table_name, "source_kind", "TEXT NOT NULL DEFAULT 'LOCAL'")
_ensure_column(conn, table_name, "views_available", "INTEGER NOT NULL DEFAULT 0")
migrations = (
(1, "initial_storage_foundation"),
(2, "unified_media_saved_flow"),
(3, "analyze8780_repository_and_10m_stats"),
(4, "collection_account_queue_sqlite_source"),
(5, "original_filename_and_account_folder_display"),
(6, "observation_sort_mode_and_hourly_central_sync"),
(7, "views_availability_and_central_hour_history"),
)
conn.executemany(
"INSERT OR IGNORE INTO schema_migrations(version, name) VALUES (?, ?)",
migrations,
)
def initialize_database(
db_path: str | Path,
st_root: str | Path,
*,
create_storage_dirs: bool = True,
) -> dict[str, Any]:
stdown_root = derive_stdown_root(st_root)
instagram_root = stdown_root / "instagram"
if create_storage_dirs:
instagram_root.mkdir(parents=True, exist_ok=True)
conn = connect_database(db_path)
try:
_apply_schema(conn)
conn.execute(
"""
INSERT INTO storage_roots(root_code, root_path, is_default)
VALUES (?, ?, 1)
ON CONFLICT(root_code) DO UPDATE SET
root_path = excluded.root_path,
is_default = 1,
updated_at = CURRENT_TIMESTAMP
""",
(DEFAULT_STORAGE_CODE, str(instagram_root)),
)
conn.execute("INSERT OR IGNORE INTO collector_profile(id) VALUES (1)")
conn.execute("INSERT OR IGNORE INTO device_profile(id) VALUES (1)")
conn.commit()
table_count = conn.execute(
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'"
).fetchone()[0]
journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0]
user_version = conn.execute("PRAGMA user_version").fetchone()[0]
return {
"db_path": str(Path(db_path).expanduser().resolve()),
"storage_root": str(instagram_root),
"schema_version": int(user_version),
"table_count": int(table_count),
"journal_mode": str(journal_mode).lower(),
"foreign_keys": int(foreign_keys),
}
finally:
conn.close()
def normalize_source_mode(source_mode: str) -> str:
normalized = str(source_mode or "").strip().upper()
if normalized not in SOURCE_MODES:
allowed = ", ".join(sorted(SOURCE_MODES))
raise ValueError(f"unsupported source_mode={source_mode!r}; allowed={allowed}")
return normalized
def canonical_instagram_post_url(shortcode: str, *candidate_urls: str, fallback_kind: str = "p") -> str:
"""Return a real Instagram post permalink, never a feed/profile scope URL."""
code = str(shortcode or "").strip()
if not code:
return ""
canonical_code = urllib.parse.quote(code, safe="-_")
kind_map = {"p": "p", "reel": "reel", "reels": "reel", "tv": "tv"}
for candidate in candidate_urls:
raw = str(candidate or "").strip()
if not raw:
continue
if raw.startswith("/"):
raw = "https://www.instagram.com" + raw
try:
parsed = urllib.parse.urlparse(raw)
except Exception:
continue
host = str(parsed.hostname or "").lower()
if host != "instagram.com" and not host.endswith(".instagram.com"):
continue
parts = [urllib.parse.unquote(part) for part in str(parsed.path or "").split("/") if part]
if len(parts) < 2:
continue
kind = kind_map.get(parts[0].lower())
if not kind or parts[1] != code:
continue
return f"https://www.instagram.com/{kind}/{canonical_code}/"
fallback = kind_map.get(str(fallback_kind or "p").strip().lower(), "p")
return f"https://www.instagram.com/{fallback}/{canonical_code}/"
def normalize_relative_path(relative_path: str | Path) -> str:
raw = str(relative_path or "").replace("\\", "/").strip()
if not raw:
raise ValueError("relative_path is required")
path = PurePosixPath(raw)
if path.is_absolute() or ".." in path.parts:
raise ValueError(f"relative_path must stay below the configured storage root: {raw}")
return path.as_posix()
def create_download_batch(
db_path: str | Path,
*,
source_mode: str,
batch_key: str | None = None,
account_id: int | None = None,
sort_by: str = "",
filter_days: int | None = None,
requested_count: int = 0,
metadata: Mapping[str, Any] | None = None,
) -> int:
mode = normalize_source_mode(source_mode)
key = str(batch_key or uuid.uuid4())
conn = connect_database(db_path)
try:
cursor = conn.execute(
"""
INSERT INTO download_batches(
batch_key, source_mode, account_id, sort_by, filter_days,
requested_count, metadata_json
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
key,
mode,
account_id,
str(sort_by or ""),
filter_days,
max(0, int(requested_count)),
json.dumps(dict(metadata or {}), ensure_ascii=False, sort_keys=True),
),
)
conn.commit()
return int(cursor.lastrowid)
finally:
conn.close()
def create_download_job(
db_path: str | Path,
*,
post_id: int,
source_mode: str,
asset_kind: str,
media_index: int = 0,
batch_id: int | None = None,
rank_no: int | None = None,
source_url: str = "",
expected_file_name: str = "",
) -> int:
mode = normalize_source_mode(source_mode)
kind = str(asset_kind or "").strip().upper()
if not kind:
raise ValueError("asset_kind is required")
conn = connect_database(db_path)
try:
cursor = conn.execute(
"""
INSERT INTO download_jobs(
batch_id, post_id, source_mode, asset_kind, media_index,
rank_no, source_url, expected_file_name
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
batch_id,
int(post_id),
mode,
kind,
max(0, int(media_index)),
rank_no,
str(source_url or ""),
str(expected_file_name or ""),
),
)
conn.commit()
return int(cursor.lastrowid)
finally:
conn.close()
def record_media_saved(
db_path: str | Path,
*,
post_id: int,
source_mode: str,
asset_kind: str,
relative_path: str | Path,
file_name: str,
folder_name: str = "",
file_size: int,
media_index: int = 0,
rank_no: int | None = None,
md5: str = "",
sha256: str = "",
source_url: str = "",
storage_root_id: int | None = None,
download_job_id: int | None = None,
event_uuid: str | None = None,
completed_at: str | None = None,
payload: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Record the common MEDIA_SAVED result for every download entry path.
The server sync payload intentionally excludes the local relative path. The
local event retains only a path relative to the configured storage root.
"""
mode = normalize_source_mode(source_mode)
kind = str(asset_kind or "").strip().upper()
if not kind:
raise ValueError("asset_kind is required")
safe_relative_path = normalize_relative_path(relative_path)
safe_file_name = str(file_name or "").strip()
if not safe_file_name:
raise ValueError("file_name is required")
relative_parent = PurePosixPath(safe_relative_path).parent
derived_folder_name = "" if str(relative_parent) in ("", ".") else relative_parent.name
safe_folder_name = str(folder_name or derived_folder_name or "instagram").strip()
size = max(0, int(file_size))
index = max(0, int(media_index))
event_key = str(event_uuid or uuid.uuid4())
completed_value = completed_at or None
conn = connect_database(db_path)
try:
conn.execute("BEGIN IMMEDIATE")
existing = conn.execute(
"SELECT id, asset_id FROM media_saved_events WHERE event_uuid = ?",
(event_key,),
).fetchone()
if existing:
conn.commit()
return {
"event_id": int(existing["id"]),
"asset_id": int(existing["asset_id"]),
"event_uuid": event_key,
"source_mode": mode,
"idempotent": True,
}
job_row = None
if download_job_id is not None:
job_row = conn.execute(
"""
SELECT id, batch_id, post_id, source_mode, asset_kind, media_index
FROM download_jobs WHERE id = ?
""",
(int(download_job_id),),
).fetchone()
if not job_row:
raise ValueError(f"download_job_id not found: {download_job_id}")
expected = (int(post_id), mode, kind, index)
actual = (
int(job_row["post_id"]),
str(job_row["source_mode"]),
str(job_row["asset_kind"]),
int(job_row["media_index"]),
)
if actual != expected:
raise ValueError(
f"download job mismatch: expected={expected!r} actual={actual!r}"
)
post = conn.execute(
"SELECT id, account_id, shortcode FROM posts WHERE id = ?",
(int(post_id),),
).fetchone()
if not post:
raise ValueError(f"post_id not found: {post_id}")
if storage_root_id is None:
default_root = conn.execute(
"SELECT id FROM storage_roots WHERE is_default = 1 ORDER BY id LIMIT 1"
).fetchone()
storage_root_id = int(default_root["id"]) if default_root else None
conn.execute(
"""
INSERT INTO media_assets(
post_id, asset_kind, media_index, source_url, storage_root_id,
relative_path, file_name, folder_name, file_size, md5, sha256,
local_status, verified_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'AVAILABLE', COALESCE(?, CURRENT_TIMESTAMP))
ON CONFLICT(post_id, asset_kind, media_index) DO UPDATE SET
source_url = CASE WHEN excluded.source_url <> '' THEN excluded.source_url ELSE media_assets.source_url END,
storage_root_id = COALESCE(excluded.storage_root_id, media_assets.storage_root_id),
relative_path = excluded.relative_path,
file_name = excluded.file_name,
folder_name = excluded.folder_name,
file_size = excluded.file_size,
md5 = excluded.md5,
sha256 = excluded.sha256,
local_status = 'AVAILABLE',
verified_at = excluded.verified_at,
updated_at = CURRENT_TIMESTAMP
""",
(
int(post_id),
kind,
index,
str(source_url or ""),
storage_root_id,
safe_relative_path,
safe_file_name,
safe_folder_name,
size,
str(md5 or ""),
str(sha256 or ""),
completed_value,
),
)
asset = conn.execute(
"SELECT id FROM media_assets WHERE post_id = ? AND asset_kind = ? AND media_index = ?",
(int(post_id), kind, index),
).fetchone()
asset_id = int(asset["id"])
local_payload = dict(payload or {})
event_cursor = conn.execute(
"""
INSERT INTO media_saved_events(
event_uuid, download_job_id, post_id, asset_id, account_id,
shortcode, source_mode, asset_kind, media_index, rank_no,
relative_path, file_name, folder_name, file_size, md5, sha256,
completed_at, payload_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), ?)
""",
(
event_key,
download_job_id,
int(post_id),
asset_id,
post["account_id"],
str(post["shortcode"]),
mode,
kind,
index,
rank_no,
safe_relative_path,
safe_file_name,
safe_folder_name,
size,
str(md5 or ""),
str(sha256 or ""),
completed_value,
json.dumps(local_payload, ensure_ascii=False, sort_keys=True),
),
)
event_id = int(event_cursor.lastrowid)
if download_job_id is not None:
conn.execute(
"""
UPDATE download_jobs
SET job_status = 'COMPLETE', completed_at = COALESCE(?, CURRENT_TIMESTAMP),
last_error = '', updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(completed_value, int(download_job_id)),
)
batch_id = job_row["batch_id"] if job_row else None
if batch_id is not None:
batch_progress = conn.execute(
"""
SELECT b.requested_count,
COUNT(j.id) AS total_jobs,
SUM(CASE WHEN j.job_status <> 'COMPLETE' THEN 1 ELSE 0 END) AS remaining_jobs
FROM download_batches b
LEFT JOIN download_jobs j ON j.batch_id = b.id
WHERE b.id = ?
GROUP BY b.id, b.requested_count
""",
(int(batch_id),),
).fetchone()
requested = int(batch_progress["requested_count"] or 0) if batch_progress else 0
total_jobs = int(batch_progress["total_jobs"] or 0) if batch_progress else 0
remaining = int(batch_progress["remaining_jobs"] or 0) if batch_progress else 0
if remaining == 0 and (requested <= 0 or total_jobs >= requested):
conn.execute(
"""
UPDATE download_batches
SET batch_status = 'COMPLETE',
completed_at = COALESCE(?, CURRENT_TIMESTAMP),
updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(completed_value, int(batch_id)),
)
else:
conn.execute(
"""
UPDATE download_batches
SET batch_status = 'RUNNING', completed_at = NULL,
updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(int(batch_id),),
)
conn.execute(
"UPDATE posts SET media_state = 'MEDIA_AVAILABLE', updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(int(post_id),),
)
conn.execute(
"""
INSERT INTO p2p_inventory(asset_id, available, updated_at)
VALUES (?, 1, CURRENT_TIMESTAMP)
ON CONFLICT(asset_id) DO UPDATE SET available = 1, updated_at = CURRENT_TIMESTAMP
""",
(asset_id,),
)
server_payload = {
"client_event_id": event_key,
"post_id": int(post_id),
"shortcode": str(post["shortcode"]),
"source_mode": mode,
"asset_kind": kind,
"media_index": index,
"rank_no": rank_no,
"file_name": safe_file_name,
"file_size": size,
"md5": str(md5 or ""),
"sha256": str(sha256 or ""),
"completed_at": completed_value,
}
conn.execute(
"""
INSERT INTO sync_queue(entity_type, entity_local_id, operation, payload_json)
VALUES ('MEDIA_SAVED', ?, 'UPSERT', ?)
""",
(event_id, json.dumps(server_payload, ensure_ascii=False, sort_keys=True)),
)
conn.commit()
return {
"event_id": event_id,
"asset_id": asset_id,
"event_uuid": event_key,
"source_mode": mode,
"folder_name": safe_folder_name,
"file_name": safe_file_name,
"idempotent": False,
}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def ensure_storage_root(
db_path: str | Path,
*,
root_path: str | Path,
root_code: str | None = None,
is_default: bool = False,
) -> int:
resolved = Path(root_path).expanduser().resolve()
code = str(root_code or "").strip()
if not code:
digest = hashlib.sha256(str(resolved).encode("utf-8")).hexdigest()[:12].upper()
code = f"LOCAL_DOWNLOAD_{digest}"
conn = connect_database(db_path)
try:
if is_default:
conn.execute("UPDATE storage_roots SET is_default = 0 WHERE is_default = 1")
conn.execute(
"""
INSERT INTO storage_roots(root_code, root_path, is_default)
VALUES (?, ?, ?)
ON CONFLICT(root_code) DO UPDATE SET
root_path = excluded.root_path,
is_default = excluded.is_default,
updated_at = CURRENT_TIMESTAMP
""",
(code, str(resolved), 1 if is_default else 0),
)
row = conn.execute("SELECT id FROM storage_roots WHERE root_code = ?", (code,)).fetchone()
conn.commit()
return int(row["id"])
finally:
conn.close()
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def resolve_download_storage(
db_path: str | Path,
*,
file_path: str | Path,
account_path: str = "",
account_folder_merge: bool = False,
) -> dict[str, Any]:
file = Path(file_path).expanduser().resolve()
if not file.is_file():
raise FileNotFoundError(str(file))
conn = connect_database(db_path)
try:
default = conn.execute(
"SELECT id, root_path FROM storage_roots WHERE is_default = 1 ORDER BY id LIMIT 1"
).fetchone()
finally:
conn.close()
if default:
default_root = Path(str(default["root_path"])).expanduser().resolve()
if _is_relative_to(file, default_root):
return {
"storage_root_id": int(default["id"]),
"storage_root_path": str(default_root),
"relative_path": file.relative_to(default_root).as_posix(),
"folder_name": file.parent.name or "instagram",
"storage_layout": "STDOWN_DEFAULT",
}
parent = file.parent
root = parent
account_parts = [part for part in str(account_path or "").replace("\\", "/").split("/") if part]
if not account_folder_merge and account_parts:
probe = parent
matched = True
for expected in reversed(account_parts):
if probe.name.casefold() != expected.casefold():
matched = False
break
probe = probe.parent
if matched:
root = probe
root_id = ensure_storage_root(db_path, root_path=root)
return {
"storage_root_id": root_id,
"storage_root_path": str(root),
"relative_path": file.relative_to(root).as_posix(),
"folder_name": file.parent.name or "instagram",
"storage_layout": "LOCAL_DOWNLOAD_ROOT",
}
def upsert_download_post(
db_path: str | Path,
*,
shortcode: str,
owner_username: str,
post_url: str = "",
caption: str = "",
thumbnail_url: str = "",
video_url: str = "",
published_at: str | None = None,
stats: Mapping[str, Any] | None = None,
fallback_kind: str = "",
) -> dict[str, int]:
code = str(shortcode or "").strip()
owner = str(owner_username or "").strip().lstrip("@")
if not code:
raise ValueError("shortcode is required")
if not owner:
raise ValueError("owner_username is required")
kind_hint = str(fallback_kind or "").strip().lower()
if kind_hint not in {"reel", "p"}:
kind_hint = "reel" if str(video_url or "").strip() else "p"
url = canonical_instagram_post_url(code, post_url, fallback_kind=kind_hint)
conn = connect_database(db_path)
try:
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"""
INSERT INTO social_accounts(platform, username, profile_url)
VALUES ('instagram', ?, ?)
ON CONFLICT(platform, username) DO UPDATE SET
profile_url = CASE WHEN excluded.profile_url <> '' THEN excluded.profile_url ELSE social_accounts.profile_url END,
updated_at = CURRENT_TIMESTAMP
""",
(owner, f"https://www.instagram.com/{owner}/"),
)
account = conn.execute(
"SELECT id FROM social_accounts WHERE platform = 'instagram' AND username = ?",
(owner,),
).fetchone()
account_id = int(account["id"])
conn.execute(
"""
INSERT INTO posts(
account_id, platform, shortcode, post_url, caption, published_at,
thumbnail_url, video_url, source_url_observed_at
) VALUES (?, 'instagram', ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(platform, shortcode) DO UPDATE SET
account_id = excluded.account_id,
post_url = CASE WHEN excluded.post_url <> '' THEN excluded.post_url ELSE posts.post_url END,
caption = CASE WHEN excluded.caption <> '' THEN excluded.caption ELSE posts.caption END,
published_at = COALESCE(excluded.published_at, posts.published_at),
thumbnail_url = CASE WHEN excluded.thumbnail_url <> '' THEN excluded.thumbnail_url ELSE posts.thumbnail_url END,
video_url = CASE WHEN excluded.video_url <> '' THEN excluded.video_url ELSE posts.video_url END,
source_url_observed_at = CURRENT_TIMESTAMP,
last_observed_at = CURRENT_TIMESTAMP,
updated_at = CURRENT_TIMESTAMP
""",
(
account_id,
code,
url,
str(caption or ""),
published_at,
str(thumbnail_url or ""),
str(video_url or ""),
),
)
post = conn.execute(
"SELECT id FROM posts WHERE platform = 'instagram' AND shortcode = ?",
(code,),
).fetchone()
post_id = int(post["id"])
values = dict(stats or {})
if values:
collected_at = str(values.get("captured_at") or values.get("collected_at") or "").strip() or None
metric_values = (
int(values.get("views") or 0),
int(values.get("likes") or 0),
int(values.get("comments") or 0),
int(values.get("reposts") or 0),
)
conn.execute(
"""
INSERT INTO post_stat_latest(post_id, views, likes, comments, reposts, collected_at)
VALUES (?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP))
ON CONFLICT(post_id) DO UPDATE SET
views = excluded.views,
likes = excluded.likes,
comments = excluded.comments,
reposts = excluded.reposts,
collected_at = excluded.collected_at
""",
(post_id, *metric_values, collected_at),
)
conn.execute(
"""
INSERT OR IGNORE INTO post_stat_history(
post_id, views, likes, comments, reposts, collected_at
) VALUES (?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP))
""",
(post_id, *metric_values, collected_at),
)
conn.commit()
return {"account_id": account_id, "post_id": post_id}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def ensure_download_batch(
db_path: str | Path,
*,
batch_key: str,
source_mode: str,
account_id: int | None = None,
sort_by: str = "",
filter_days: int | None = None,
requested_count: int = 0,
metadata: Mapping[str, Any] | None = None,
) -> int:
key = str(batch_key or "").strip()
if not key:
raise ValueError("batch_key is required")
mode = normalize_source_mode(source_mode)
conn = connect_database(db_path)
try:
conn.execute(
"""
INSERT INTO download_batches(
batch_key, source_mode, account_id, sort_by, filter_days,
requested_count, batch_status, metadata_json
) VALUES (?, ?, ?, ?, ?, ?, 'RUNNING', ?)
ON CONFLICT(batch_key) DO UPDATE SET
account_id = COALESCE(excluded.account_id, download_batches.account_id),
sort_by = CASE WHEN excluded.sort_by <> '' THEN excluded.sort_by ELSE download_batches.sort_by END,
filter_days = COALESCE(excluded.filter_days, download_batches.filter_days),
requested_count = MAX(download_batches.requested_count, excluded.requested_count),
batch_status = CASE WHEN download_batches.batch_status = 'COMPLETE' THEN 'COMPLETE' ELSE 'RUNNING' END,
metadata_json = excluded.metadata_json,
updated_at = CURRENT_TIMESTAMP
""",
(
key,
mode,
account_id,
str(sort_by or ""),
filter_days,
max(0, int(requested_count)),
json.dumps(dict(metadata or {}), ensure_ascii=False, sort_keys=True),
),
)
row = conn.execute("SELECT id FROM download_batches WHERE batch_key = ?", (key,)).fetchone()
conn.commit()
return int(row["id"])
finally:
conn.close()
def ensure_download_job(
db_path: str | Path,
*,
batch_id: int,
post_id: int,
source_mode: str,
asset_kind: str,
media_index: int = 0,
rank_no: int | None = None,
source_url: str = "",
expected_file_name: str = "",
) -> int:
mode = normalize_source_mode(source_mode)
kind = str(asset_kind or "").strip().upper()
if not kind:
raise ValueError("asset_kind is required")
conn = connect_database(db_path)
try:
conn.execute(
"""
INSERT INTO download_jobs(
batch_id, post_id, source_mode, asset_kind, media_index,
rank_no, source_url, expected_file_name, job_status
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'DOWNLOADING')
ON CONFLICT(batch_id, post_id, asset_kind, media_index) DO UPDATE SET
rank_no = COALESCE(excluded.rank_no, download_jobs.rank_no),
source_url = CASE WHEN excluded.source_url <> '' THEN excluded.source_url ELSE download_jobs.source_url END,
expected_file_name = CASE WHEN excluded.expected_file_name <> '' THEN excluded.expected_file_name ELSE download_jobs.expected_file_name END,
job_status = CASE WHEN download_jobs.job_status = 'COMPLETE' THEN 'COMPLETE' ELSE 'DOWNLOADING' END,
last_error = '',
updated_at = CURRENT_TIMESTAMP
""",
(
int(batch_id),
int(post_id),
mode,
kind,
max(0, int(media_index)),
rank_no,
str(source_url or ""),
str(expected_file_name or ""),
),
)
row = conn.execute(
"""
SELECT id FROM download_jobs
WHERE batch_id = ? AND post_id = ? AND asset_kind = ? AND media_index = ?
""",
(int(batch_id), int(post_id), kind, max(0, int(media_index))),
).fetchone()
conn.commit()
return int(row["id"])
finally:
conn.close()
def record_downloaded_file(
*,
st_root: str | Path,
file_path: str | Path,
db_path: str | Path | None = None,
source_mode: str,
shortcode: str,
owner_username: str,
asset_kind: str,
media_index: int = 0,
rank_no: int | None = None,
source_url: str = "",
post_url: str = "",
caption: str = "",
thumbnail_url: str = "",
video_url: str = "",
published_at: str | None = None,
md5: str = "",
sha256: str = "",
file_size: int = 0,
account_path: str = "",
account_folder_merge: bool = False,
batch_key: str = "",
requested_count: int = 0,
sort_by: str = "",
filter_days: int | None = None,
event_uuid: str = "",
stats: Mapping[str, Any] | None = None,
payload: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
root = Path(st_root).expanduser().resolve()
resolved_db_path = Path(db_path).expanduser().resolve() if db_path else default_database_path(root)
initialize_database(resolved_db_path, root)
mode = normalize_source_mode(source_mode)
event_key = str(event_uuid or uuid.uuid4())
context = upsert_download_post(
resolved_db_path,
shortcode=shortcode,
owner_username=owner_username,
post_url=post_url,
caption=caption,
thumbnail_url=thumbnail_url,
video_url=video_url,
published_at=published_at,
stats=stats,
fallback_kind="reel" if str(asset_kind or "").strip().upper() == "VIDEO" else "p",
)
storage = resolve_download_storage(
resolved_db_path,
file_path=file_path,
account_path=account_path,
account_folder_merge=account_folder_merge,
)
effective_batch_key = str(batch_key or f"single-{event_key}")
batch_id = ensure_download_batch(
resolved_db_path,
batch_key=effective_batch_key,
source_mode=mode,
account_id=context["account_id"],
sort_by=sort_by,
filter_days=filter_days,
requested_count=max(1, int(requested_count or 0)),
metadata={
"account_path": str(account_path or ""),
"storage_layout": storage["storage_layout"],
},
)
job_id = ensure_download_job(
resolved_db_path,
batch_id=batch_id,
post_id=context["post_id"],
source_mode=mode,
asset_kind=asset_kind,
media_index=media_index,
rank_no=rank_no,
source_url=source_url,
expected_file_name=Path(file_path).name,
)
local_payload = dict(payload or {})
local_payload.update(
{
"batch_key": effective_batch_key,
"storage_layout": storage["storage_layout"],
}
)
try:
result = record_media_saved(
resolved_db_path,
post_id=context["post_id"],
source_mode=mode,
asset_kind=asset_kind,
relative_path=storage["relative_path"],
file_name=Path(file_path).name,
folder_name=storage["folder_name"],
file_size=file_size,
media_index=media_index,
rank_no=rank_no,
md5=md5,
sha256=sha256,
source_url=source_url,
storage_root_id=storage["storage_root_id"],
download_job_id=job_id,
event_uuid=event_key,
payload=local_payload,
)
except Exception as exc:
conn = connect_database(db_path)
try:
conn.execute(
"""
UPDATE download_jobs
SET job_status = 'FAILED', last_error = ?, updated_at = CURRENT_TIMESTAMP
WHERE id = ?
""",
(str(exc), job_id),
)
conn.commit()
finally:
conn.close()
raise
return {
**result,
"db_path": str(resolved_db_path),
"account_id": context["account_id"],
"post_id": context["post_id"],
"batch_id": batch_id,
"job_id": job_id,
"relative_path": storage["relative_path"],
"folder_name": storage["folder_name"],
"file_name": Path(file_path).name,
"storage_root_id": storage["storage_root_id"],
"storage_layout": storage["storage_layout"],
}
def _parse_observed_at(value: str | datetime | None) -> datetime:
if isinstance(value, datetime):
dt = value
else:
raw = str(value or "").strip()
if not raw:
dt = datetime.now(timezone.utc)
else:
if raw.endswith("Z"):
raw = raw[:-1] + "+00:00"
dt = datetime.fromisoformat(raw)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
def ten_minute_bucket(value: str | datetime | None = None) -> str:
dt = _parse_observed_at(value)
minute = (dt.minute // 10) * 10
return dt.replace(minute=minute, second=0, microsecond=0).isoformat().replace("+00:00", "Z")
def _iso_utc(value: str | datetime | None = None) -> str:
return _parse_observed_at(value).isoformat().replace("+00:00", "Z")
def upsert_observation(
db_path: str | Path,
*,
username: str,
shortcode: str,
views: int = 0,
views_available: int = 0,
likes: int = 0,
comments: int = 0,
reposts: int = 0,
observed_at: str | datetime | None = None,
platform_account_id: str = "",
display_name: str = "",
profile_url: str = "",
post_url: str = "",
caption: str = "",
published_at: str | None = None,
thumbnail_url: str = "",
video_url: str = "",
sort_by: str = "",
sort_value: int = 0,
source_kind: str = "LOCAL",
) -> dict[str, Any]:
owner = str(username or "").strip().lstrip("@")
code = str(shortcode or "").strip()
if not owner:
raise ValueError("username is required")
if not code:
raise ValueError("shortcode is required")
observed = _iso_utc(observed_at)
bucket = ten_minute_bucket(observed_at)
metrics = tuple(max(0, int(v or 0)) for v in (views, likes, comments, reposts))
views_available_flag = 1 if int(views_available or 0) else 0
sort_field = str(sort_by or "").strip()
sort_metric_value = max(0, int(sort_value or 0))
observation_source = str(source_kind or "LOCAL").strip().upper() or "LOCAL"
conn = connect_database(db_path)
try:
conn.execute("BEGIN IMMEDIATE")
deleted_account = conn.execute(
"SELECT 1 FROM deletion_events WHERE entity_type='ACCOUNT' AND lower(entity_key)=lower(?) LIMIT 1",
(owner,),
).fetchone()
deleted_post = conn.execute(
"SELECT 1 FROM deletion_events WHERE entity_type='POST' AND entity_key=? LIMIT 1",
(code,),
).fetchone()
if deleted_account or deleted_post:
conn.commit()
return {
"account_id": 0,
"post_id": 0,
"bucket_at": bucket,
"observed_at": observed,
"ignored_deleted": True,
"deleted_entity": "ACCOUNT" if deleted_account else "POST",
}
existing_observation = conn.execute(
"""
SELECT
p.id AS post_id, p.account_id, p.post_url, p.caption, p.published_at,
p.thumbnail_url, p.video_url, p.deleted_at AS post_deleted_at,
a.username, a.platform_account_id, a.display_name, a.profile_url,
a.active, a.deleted_at AS account_deleted_at,
s.views, s.views_available, s.likes, s.comments, s.reposts
FROM posts p
JOIN social_accounts a ON a.id=p.account_id
LEFT JOIN post_stat_latest s ON s.post_id=p.id
WHERE p.platform='instagram' AND p.shortcode=?
LIMIT 1
""",
(code,),
).fetchone()
if existing_observation is not None and existing_observation["views"] is not None:
effective_views = metrics[0]
effective_views_available = views_available_flag
if not views_available_flag and int(existing_observation["views_available"] or 0):
effective_views = max(0, int(existing_observation["views"] or 0))
effective_views_available = 1
unchanged_metrics = (
int(existing_observation["views"] or 0),
int(existing_observation["views_available"] or 0),
int(existing_observation["likes"] or 0),
int(existing_observation["comments"] or 0),
int(existing_observation["reposts"] or 0),
) == (
effective_views, effective_views_available, metrics[1], metrics[2], metrics[3]
)
incoming_post_url = str(post_url or "").strip()
resolved_incoming_post_url = (
canonical_instagram_post_url(
code, incoming_post_url,
fallback_kind="reel" if str(video_url or "").strip() else "p",
)
if incoming_post_url else ""
)
metadata_changed = any((
str(existing_observation["username"] or "").strip().lower() != owner.lower(),
int(existing_observation["active"] or 0) != 1,
existing_observation["account_deleted_at"] is not None,
existing_observation["post_deleted_at"] is not None,
bool(str(platform_account_id or "").strip()) and str(existing_observation["platform_account_id"] or "") != str(platform_account_id or ""),
bool(str(display_name or "").strip()) and str(existing_observation["display_name"] or "") != str(display_name or ""),
bool(str(profile_url or "").strip()) and str(existing_observation["profile_url"] or "") != str(profile_url or ""),
bool(resolved_incoming_post_url) and canonical_instagram_post_url(code, existing_observation["post_url"], fallback_kind="reel" if str(existing_observation["video_url"] or "").strip() else "p") != resolved_incoming_post_url,
bool(str(caption or "")) and str(existing_observation["caption"] or "") != str(caption or ""),
published_at is not None and str(existing_observation["published_at"] or "") != str(published_at or ""),
bool(str(thumbnail_url or "").strip()) and str(existing_observation["thumbnail_url"] or "") != str(thumbnail_url or ""),
bool(str(video_url or "").strip()) and str(existing_observation["video_url"] or "") != str(video_url or ""),
))
if unchanged_metrics and not metadata_changed:
conn.commit()
return {
"account_id": int(existing_observation["account_id"]),
"post_id": int(existing_observation["post_id"]),
"bucket_at": bucket,
"observed_at": observed,
"views_available": effective_views_available,
"sort_by": sort_field,
"sort_value": sort_metric_value,
"source_kind": observation_source,
"saved": False,
"skipped_unchanged": True,
"skipped_all_writes": True,
}
conn.execute(
"""
INSERT INTO social_accounts(
platform, platform_account_id, username, display_name, profile_url, active, deleted_at
) VALUES ('instagram', ?, ?, ?, ?, 1, NULL)
ON CONFLICT(platform, username) DO UPDATE SET
platform_account_id = CASE WHEN excluded.platform_account_id <> '' THEN excluded.platform_account_id ELSE social_accounts.platform_account_id END,
display_name = CASE WHEN excluded.display_name <> '' THEN excluded.display_name ELSE social_accounts.display_name END,
profile_url = CASE WHEN excluded.profile_url <> '' THEN excluded.profile_url ELSE social_accounts.profile_url END,
active = 1,
deleted_at = NULL,
updated_at = CURRENT_TIMESTAMP
""",
(str(platform_account_id or ""), owner, str(display_name or ""), str(profile_url or "") or f"https://www.instagram.com/{owner}/"),
)
account_id = int(conn.execute(
"SELECT id FROM social_accounts WHERE platform='instagram' AND username=?", (owner,)
).fetchone()["id"])
existing_post = conn.execute(
"SELECT post_url, video_url FROM posts WHERE platform='instagram' AND shortcode=?",
(code,),
).fetchone()
if existing_post and not str(post_url or "").strip() and not str(video_url or "").strip():
resolved_post_url = canonical_instagram_post_url(
code,
existing_post["post_url"],
fallback_kind="reel" if str(existing_post["video_url"] or "").strip() else "p",
)
else:
resolved_post_url = canonical_instagram_post_url(
code,
post_url,
fallback_kind="reel" if str(video_url or "").strip() else "p",
)
conn.execute(
"""
INSERT INTO posts(
account_id, platform, shortcode, post_url, caption, published_at,
thumbnail_url, video_url, source_url_observed_at, last_observed_at, deleted_at
) VALUES (?, 'instagram', ?, ?, ?, ?, ?, ?, ?, ?, NULL)
ON CONFLICT(platform, shortcode) DO UPDATE SET
account_id = excluded.account_id,
post_url = CASE WHEN excluded.post_url <> '' THEN excluded.post_url ELSE posts.post_url END,
caption = CASE WHEN excluded.caption <> '' THEN excluded.caption ELSE posts.caption END,
published_at = COALESCE(excluded.published_at, posts.published_at),
thumbnail_url = CASE WHEN excluded.thumbnail_url <> '' THEN excluded.thumbnail_url ELSE posts.thumbnail_url END,
video_url = CASE WHEN excluded.video_url <> '' THEN excluded.video_url ELSE posts.video_url END,
source_url_observed_at = excluded.source_url_observed_at,
last_observed_at = excluded.last_observed_at,
deleted_at = NULL,
updated_at = CURRENT_TIMESTAMP
""",
(
account_id, code, resolved_post_url,
str(caption or ""), published_at, str(thumbnail_url or ""), str(video_url or ""),
observed, observed,
),
)
post_id = int(conn.execute(
"SELECT id FROM posts WHERE platform='instagram' AND shortcode=?", (code,)
).fetchone()["id"])
previous_stat = conn.execute(
"""
SELECT views, views_available, likes, comments, reposts
FROM post_stat_latest
WHERE post_id=?
""",
(post_id,),
).fetchone()
effective_views = metrics[0]
effective_views_available = views_available_flag
if previous_stat is not None and not views_available_flag and int(previous_stat["views_available"] or 0):
effective_views = max(0, int(previous_stat["views"] or 0))
effective_views_available = 1
effective_metrics = (effective_views, metrics[1], metrics[2], metrics[3])
unchanged = previous_stat is not None and (
int(previous_stat["views"] or 0),
int(previous_stat["views_available"] or 0),
int(previous_stat["likes"] or 0),
int(previous_stat["comments"] or 0),
int(previous_stat["reposts"] or 0),
) == (
effective_metrics[0],
effective_views_available,
effective_metrics[1],
effective_metrics[2],
effective_metrics[3],
)
if unchanged:
conn.execute(
"DELETE FROM deletion_events WHERE entity_type='ACCOUNT' AND entity_local_id=?", (account_id,)
)
conn.execute(
"DELETE FROM deletion_events WHERE entity_type='POST' AND entity_local_id=?", (post_id,)
)
conn.commit()
return {
"account_id": account_id,
"post_id": post_id,
"bucket_at": bucket,
"observed_at": observed,
"views_available": effective_views_available,
"sort_by": sort_field,
"sort_value": sort_metric_value,
"source_kind": observation_source,
"saved": False,
"skipped_unchanged": True,
}
conn.execute(
"""
INSERT INTO post_stat_latest(post_id, views, views_available, likes, comments, reposts, collected_at, sort_by, sort_value, source_kind)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(post_id) DO UPDATE SET
views=excluded.views, views_available=excluded.views_available, likes=excluded.likes, comments=excluded.comments,
reposts=excluded.reposts, collected_at=excluded.collected_at,
sort_by=excluded.sort_by, sort_value=excluded.sort_value, source_kind=excluded.source_kind
""",
(post_id, effective_metrics[0], effective_views_available, effective_metrics[1], effective_metrics[2], effective_metrics[3], observed, sort_field, sort_metric_value, observation_source),
)
conn.execute(
"""
INSERT INTO post_stat_history(post_id, views, views_available, likes, comments, reposts, collected_at, sort_by, sort_value, source_kind)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(post_id, collected_at) DO UPDATE SET
views=excluded.views, views_available=excluded.views_available, likes=excluded.likes, comments=excluded.comments, reposts=excluded.reposts,
sort_by=excluded.sort_by, sort_value=excluded.sort_value, source_kind=excluded.source_kind
""",
(post_id, effective_metrics[0], effective_views_available, effective_metrics[1], effective_metrics[2], effective_metrics[3], observed, sort_field, sort_metric_value, observation_source),
)
conn.execute(
"""
INSERT INTO post_stat_10m(
post_id, bucket_at, observed_at, views, views_available, likes, comments, reposts, sample_count, sort_by, sort_value, source_kind
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
ON CONFLICT(post_id, bucket_at) DO UPDATE SET
observed_at=excluded.observed_at, views=excluded.views, views_available=excluded.views_available, likes=excluded.likes,
comments=excluded.comments, reposts=excluded.reposts, sort_by=excluded.sort_by,
sort_value=excluded.sort_value, source_kind=excluded.source_kind,
sample_count=post_stat_10m.sample_count + 1, updated_at=CURRENT_TIMESTAMP
""",
(post_id, bucket, observed, effective_metrics[0], effective_views_available, effective_metrics[1], effective_metrics[2], effective_metrics[3], sort_field, sort_metric_value, observation_source),
)
conn.execute(
"DELETE FROM deletion_events WHERE entity_type='ACCOUNT' AND entity_local_id=?", (account_id,)
)
conn.execute(
"DELETE FROM deletion_events WHERE entity_type='POST' AND entity_local_id=?", (post_id,)
)
conn.commit()
return {"account_id": account_id, "post_id": post_id, "bucket_at": bucket, "observed_at": observed, "views_available": effective_views_available, "sort_by": sort_field, "sort_value": sort_metric_value, "source_kind": observation_source, "saved": True, "skipped_unchanged": False}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def list_accounts(
db_path: str | Path, *, page: int = 1, page_size: int = 50, include_deleted: bool = False,
search_pattern: str = "",
) -> dict[str, Any]:
page = max(1, int(page or 1))
page_size = max(1, min(10000, int(page_size or 50)))
base_where = "1=1" if include_deleted else "a.active=1 AND a.deleted_at IS NULL"
pattern = str(search_pattern or "")
where = base_where
where_params: list[Any] = []
if pattern:
where += " AND a.username LIKE ? COLLATE NOCASE"
where_params.append(pattern)
conn = connect_database(db_path)
try:
total_all = int(conn.execute(
f"SELECT COUNT(*) AS n FROM social_accounts a WHERE {base_where}"
).fetchone()["n"])
total = int(conn.execute(
f"SELECT COUNT(*) AS n FROM social_accounts a WHERE {where}", where_params
).fetchone()["n"])
rows = conn.execute(
f"""
SELECT a.id, a.username, a.display_name, a.profile_url, a.active, a.deleted_at,
COUNT(p.id) AS post_count,
COALESCE(MAX(s.views), 0) AS max_views,
COALESCE(SUM(s.views), 0) AS total_views,
MAX(COALESCE(s.collected_at, p.last_observed_at, a.updated_at)) AS last_observed_at
FROM social_accounts a
LEFT JOIN posts p ON p.account_id=a.id AND p.deleted_at IS NULL
LEFT JOIN post_stat_latest s ON s.post_id=p.id
WHERE {where}
GROUP BY a.id
ORDER BY COALESCE(last_observed_at, a.updated_at) DESC, a.username COLLATE NOCASE
LIMIT ? OFFSET ?
""",
(*where_params, page_size, (page - 1) * page_size),
).fetchall()
return {
"items": [dict(row) for row in rows],
"page": page,
"page_size": page_size,
"total": total,
"total_all": total_all,
"search_pattern": pattern,
"table_name": "social_accounts",
}
finally:
conn.close()
def list_account_posts(
db_path: str | Path, *, account_id: int, page: int = 1, page_size: int = 50, sort_by: str = "views",
include_media_assets: bool = True,
) -> dict[str, Any]:
page = max(1, int(page or 1))
page_size = max(1, min(10000, int(page_size or 50)))
sort_map = {
"views": "views DESC", "likes": "likes DESC", "comments": "comments DESC",
"reposts": "reposts DESC", "views_delta": "views_delta DESC",
"views_per_hour": "views_per_hour DESC", "published_at": "p.published_at DESC",
}
order = sort_map.get(str(sort_by or "views"), sort_map["views"])
media_columns = ""
if include_media_assets:
media_columns = """,
(SELECT ma.folder_name FROM media_assets ma WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE' ORDER BY ma.media_index, ma.id LIMIT 1) AS folder_name,
(SELECT ma.file_name FROM media_assets ma WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE' ORDER BY ma.media_index, ma.id LIMIT 1) AS file_name,
(SELECT ma.relative_path FROM media_assets ma WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE' ORDER BY ma.media_index, ma.id LIMIT 1) AS relative_path,
EXISTS(
SELECT 1 FROM media_assets ma
WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE'
AND UPPER(ma.asset_kind) IN ('IMAGE','JPG','JPEG')
) AS has_jpg,
EXISTS(
SELECT 1 FROM media_assets ma
WHERE ma.post_id=p.id AND ma.local_status='AVAILABLE'
AND UPPER(ma.asset_kind) IN ('VIDEO','MP4')
) AS has_mp4"""
conn = connect_database(db_path)
try:
total = int(conn.execute(
"SELECT COUNT(*) AS n FROM posts WHERE account_id=? AND deleted_at IS NULL", (int(account_id),)
).fetchone()["n"])
rows = conn.execute(
f"""
WITH ranked AS (
SELECT t.*, ROW_NUMBER() OVER(PARTITION BY t.post_id ORDER BY t.bucket_at DESC) AS rn
FROM post_stat_10m t
), recent AS (
SELECT post_id,
MAX(CASE WHEN rn=1 THEN views END) AS last_views,
MAX(CASE WHEN rn=2 THEN views END) AS prev_views,
MAX(CASE WHEN rn=1 THEN bucket_at END) AS bucket_at
FROM ranked WHERE rn <= 2 GROUP BY post_id
)
SELECT p.id, p.shortcode, p.post_url, p.caption, p.published_at, p.thumbnail_url,
p.video_url, p.media_state, p.last_observed_at,
COALESCE(s.views,0) AS views, COALESCE(s.views_available,0) AS views_available, COALESCE(s.likes,0) AS likes,
COALESCE(s.comments,0) AS comments, COALESCE(s.reposts,0) AS reposts,
COALESCE(recent.last_views, s.views, 0)-COALESCE(recent.prev_views, recent.last_views, s.views, 0) AS views_delta,
(COALESCE(recent.last_views, s.views, 0)-COALESCE(recent.prev_views, recent.last_views, s.views, 0))*6 AS views_per_hour,
CASE WHEN COALESCE(s.views,0)>0 THEN ROUND(COALESCE(s.likes,0)*100.0/s.views,4) ELSE 0 END AS like_rate,
CASE WHEN COALESCE(s.views,0)>0 THEN ROUND(COALESCE(s.comments,0)*100.0/s.views,4) ELSE 0 END AS comment_rate,
recent.bucket_at AS latest_bucket_at
{media_columns}
FROM posts p
LEFT JOIN post_stat_latest s ON s.post_id=p.id
LEFT JOIN recent ON recent.post_id=p.id
WHERE p.account_id=? AND p.deleted_at IS NULL
ORDER BY {order}, p.id DESC
LIMIT ? OFFSET ?
""",
(int(account_id), page_size, (page - 1) * page_size),
).fetchall()
items = [dict(row) for row in rows]
assets_by_post: dict[int, list[dict[str, Any]]] = {}
if include_media_assets:
post_ids = [int(item["id"]) for item in items]
if post_ids:
placeholders = ",".join("?" for _ in post_ids)
asset_rows = conn.execute(
f"""
SELECT ma.id, ma.post_id, ma.asset_kind, ma.media_index, ma.storage_root_id,
ma.folder_name, ma.file_name, ma.relative_path, ma.local_status, sr.root_path
FROM media_assets ma
LEFT JOIN storage_roots sr ON sr.id=ma.storage_root_id
WHERE ma.post_id IN ({placeholders}) AND ma.local_status='AVAILABLE'
ORDER BY ma.post_id, ma.media_index, ma.id
""",
post_ids,
).fetchall()
for asset_row in asset_rows:
asset = dict(asset_row)
assets_by_post.setdefault(int(asset.pop("post_id")), []).append(asset)
for item in items:
item["post_url"] = canonical_instagram_post_url(item.get("shortcode"), item.get("post_url"), fallback_kind="reel" if str(item.get("video_url") or "").strip() else "p")
if include_media_assets:
item["media_assets"] = assets_by_post.get(int(item["id"]), [])
return {"items": items, "page": page, "page_size": page_size, "total": total, "sort_by": sort_by}
finally:
conn.close()
def get_post_detail(
db_path: str | Path, post_id: int, *, include_media_assets: bool = True
) -> dict[str, Any] | None:
conn = connect_database(db_path)
try:
row = conn.execute(
"""
SELECT p.*, a.username, a.display_name,
COALESCE(s.views,0) AS views, COALESCE(s.views_available,0) AS views_available, COALESCE(s.likes,0) AS likes,
COALESCE(s.comments,0) AS comments, COALESCE(s.reposts,0) AS reposts,
s.collected_at
FROM posts p
LEFT JOIN social_accounts a ON a.id=p.account_id
LEFT JOIN post_stat_latest s ON s.post_id=p.id
WHERE p.id=? AND p.deleted_at IS NULL
""",
(int(post_id),),
).fetchone()
if not row:
return None
item = dict(row)
item["post_url"] = canonical_instagram_post_url(item.get("shortcode"), item.get("post_url"), fallback_kind="reel" if str(item.get("video_url") or "").strip() else "p")
if include_media_assets:
assets = conn.execute(
"""
SELECT id, asset_kind, media_index, storage_root_id, folder_name, file_name, relative_path, file_size, md5, sha256, local_status
FROM media_assets WHERE post_id=? ORDER BY media_index, id
""",
(int(post_id),),
).fetchall()
item["media_assets"] = [dict(asset) for asset in assets]
return item
finally:
conn.close()
def get_post_stats(db_path: str | Path, post_id: int, limit: int = 144) -> list[dict[str, Any]]:
conn = connect_database(db_path)
try:
rows = conn.execute(
"""
SELECT bucket_at, observed_at, views, views_available, likes, comments, reposts, sample_count
FROM post_stat_10m WHERE post_id=? ORDER BY bucket_at DESC LIMIT ?
""",
(int(post_id), max(1, min(1008, int(limit or 144)))),
).fetchall()
return [dict(row) for row in rows]
finally:
conn.close()
_COLLECTION_RESERVED_USERNAMES = frozenset({
"", "p", "reel", "reels", "explore", "explorer", "stories", "accounts",
"direct", "graphql", "api", "oauth", "developer", "about", "privacy",
"terms", "legal", "emails", "challenge", "search", "home", "instagram",
})
def normalize_collection_username(value: object) -> str:
"""Return a canonical Instagram username for the collection queue.
The queue stores usernames only. A legacy profile URL is accepted only for
migration; feed/detail scopes such as /explore/, /reels/, /p/ and /reel/
are rejected and never become collection accounts.
"""
raw = str(value or "").strip().replace("🗑️", "", 1).strip().lstrip("@")
if not raw:
return ""
candidate = raw
if "://" in raw or raw.startswith("www.") or raw.startswith("instagram.com/"):
url = raw if "://" in raw else "https://" + raw
try:
parsed = urllib.parse.urlparse(url)
except Exception:
return ""
host = str(parsed.hostname or "").lower().removeprefix("www.")
if host != "instagram.com":
return ""
parts = [urllib.parse.unquote(part) for part in str(parsed.path or "").split("/") if part]
if not parts:
return ""
first = str(parts[0] or "").strip()
if first.lower() in _COLLECTION_RESERVED_USERNAMES:
return ""
# Only a profile root or its own reels tab is a legacy account input.
if len(parts) > 1 and str(parts[1] or "").lower() not in {"reels"}:
return ""
candidate = first
elif "/" in raw:
return ""
candidate = candidate.strip().lstrip("@").lower()
if candidate in _COLLECTION_RESERVED_USERNAMES:
return ""
if not candidate or len(candidate) > 30:
return ""
if any(ch not in "abcdefghijklmnopqrstuvwxyz0123456789._" for ch in candidate):
return ""
return candidate
def list_collection_accounts(db_path: str | Path, *, include_deleted: bool = True) -> list[dict[str, Any]]:
conn = connect_database(db_path)
try:
where = "" if include_deleted else "WHERE enabled=1 AND is_deleted=0"
rows = conn.execute(
f"""
SELECT id, username, position, enabled, is_deleted, source_type, updated_at
FROM collection_account_queue
{where}
ORDER BY position ASC, id ASC
"""
).fetchall()
return [
{
**dict(row),
"enabled": bool(row["enabled"]),
"is_deleted": bool(row["is_deleted"]),
"account_url": f"https://www.instagram.com/{row['username']}/",
}
for row in rows
]
finally:
conn.close()
def replace_collection_accounts(db_path: str | Path, items: list[object]) -> dict[str, Any]:
"""Replace the SIDE collection queue with a username-only ordered list."""
normalized_by_username: dict[str, dict[str, Any]] = {}
for raw_item in list(items or []):
item = raw_item if isinstance(raw_item, Mapping) else {"username": raw_item}
username = normalize_collection_username(
item.get("username") or item.get("account") or item.get("account_url") or item.get("url")
)
if not username:
continue
deleted = 1 if bool(item.get("is_deleted") or item.get("deleted")) else 0
enabled = 0 if deleted else (1 if bool(item.get("enabled", True)) else 0)
previous = normalized_by_username.get(username)
if previous is not None:
previous["is_deleted"] = 1 if previous["is_deleted"] or deleted else 0
previous["enabled"] = 0 if previous["is_deleted"] else (1 if previous["enabled"] and enabled else 0)
continue
normalized_by_username[username] = {
"username": username,
"position": 0,
"enabled": enabled,
"is_deleted": deleted,
}
# Preserve the exact user-visible order received from Side. Sorting is an
# explicit Side action; a plain save must never reorder the queue.
normalized = list(normalized_by_username.values())
seen = set(normalized_by_username)
for position, item in enumerate(normalized, start=1):
item["position"] = position
conn = connect_database(db_path)
try:
conn.execute("BEGIN IMMEDIATE")
deleted_names = {
str(row["entity_key"] or "").strip().lower()
for row in conn.execute(
"SELECT entity_key FROM deletion_events WHERE entity_type='ACCOUNT'"
).fetchall()
}
for item in normalized:
if item["username"] in deleted_names:
item["is_deleted"] = 1
item["enabled"] = 0
conn.execute(
"""
INSERT INTO collection_account_queue(username, position, enabled, is_deleted, source_type)
VALUES (?, ?, ?, ?, 'SIDE')
ON CONFLICT(username) DO UPDATE SET
position=excluded.position,
enabled=excluded.enabled,
is_deleted=excluded.is_deleted,
source_type='SIDE',
updated_at=CURRENT_TIMESTAMP
""",
(item["username"], item["position"], item["enabled"], item["is_deleted"]),
)
if seen:
placeholders = ",".join("?" for _ in seen)
conn.execute(
f"DELETE FROM collection_account_queue WHERE source_type='SIDE' AND username NOT IN ({placeholders})",
tuple(sorted(seen)),
)
else:
conn.execute("DELETE FROM collection_account_queue WHERE source_type='SIDE'")
payload = {
"items": [
{
"username": item["username"],
"position": item["position"],
"enabled": bool(item["enabled"]),
"is_deleted": bool(item["is_deleted"]),
}
for item in normalized
]
}
conn.execute(
"INSERT INTO sync_queue(entity_type, operation, payload_json) VALUES('COLLECTION_ACCOUNT_QUEUE', 'REPLACE', ?)",
(json.dumps(payload, ensure_ascii=False, sort_keys=True),),
)
conn.commit()
return {"items": list_collection_accounts(db_path), "total": len(normalized)}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def soft_delete_account(db_path: str | Path, account_id: int, reason: str = "user_delete") -> dict[str, Any]:
conn = connect_database(db_path)
try:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute("SELECT id, username FROM social_accounts WHERE id=?", (int(account_id),)).fetchone()
if not row:
raise ValueError(f"account_id not found: {account_id}")
conn.execute("UPDATE social_accounts SET active=0, deleted_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP WHERE id=?", (int(account_id),))
conn.execute("UPDATE account_collection_rules SET enabled=0, updated_at=CURRENT_TIMESTAMP WHERE account_id=?", (int(account_id),))
conn.execute("UPDATE collection_account_queue SET enabled=0, is_deleted=1, updated_at=CURRENT_TIMESTAMP WHERE username=? COLLATE NOCASE", (str(row["username"]),))
conn.execute("UPDATE posts SET deleted_at=COALESCE(deleted_at,CURRENT_TIMESTAMP), updated_at=CURRENT_TIMESTAMP WHERE account_id=?", (int(account_id),))
conn.execute(
"""INSERT INTO deletion_events(entity_type, entity_local_id, entity_key, reason)
VALUES('ACCOUNT', ?, ?, ?)
ON CONFLICT(entity_type, entity_local_id) DO UPDATE SET
entity_key=excluded.entity_key, reason=excluded.reason, created_at=CURRENT_TIMESTAMP, synced_at=NULL""",
(int(account_id), str(row["username"]), str(reason or "user_delete")),
)
payload = json.dumps({"account_id": int(account_id), "username": str(row["username"]), "deleted": True}, ensure_ascii=False, sort_keys=True)
conn.execute("INSERT INTO sync_queue(entity_type, entity_local_id, operation, payload_json) VALUES('ACCOUNT', ?, 'DELETE', ?)", (int(account_id), payload))
conn.commit()
return {"account_id": int(account_id), "username": str(row["username"]), "deleted": True}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def soft_delete_post(db_path: str | Path, post_id: int, reason: str = "user_delete") -> dict[str, Any]:
conn = connect_database(db_path)
try:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute("SELECT id, shortcode FROM posts WHERE id=?", (int(post_id),)).fetchone()
if not row:
raise ValueError(f"post_id not found: {post_id}")
conn.execute("UPDATE posts SET deleted_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP WHERE id=?", (int(post_id),))
conn.execute(
"""INSERT INTO deletion_events(entity_type, entity_local_id, entity_key, reason)
VALUES('POST', ?, ?, ?)
ON CONFLICT(entity_type, entity_local_id) DO UPDATE SET
entity_key=excluded.entity_key, reason=excluded.reason, created_at=CURRENT_TIMESTAMP, synced_at=NULL""",
(int(post_id), str(row["shortcode"]), str(reason or "user_delete")),
)
payload = json.dumps({"post_id": int(post_id), "shortcode": str(row["shortcode"]), "deleted": True}, ensure_ascii=False, sort_keys=True)
conn.execute("INSERT INTO sync_queue(entity_type, entity_local_id, operation, payload_json) VALUES('POST', ?, 'DELETE', ?)", (int(post_id), payload))
conn.commit()
return {"post_id": int(post_id), "shortcode": str(row["shortcode"]), "deleted": True}
except Exception:
conn.rollback()
raise
finally:
conn.close()
def list_deleted_accounts(db_path: str | Path) -> list[dict[str, Any]]:
conn = connect_database(db_path)
try:
rows = conn.execute(
"""SELECT entity_local_id AS account_id, entity_key AS username, created_at
FROM deletion_events WHERE entity_type='ACCOUNT' ORDER BY created_at DESC"""
).fetchall()
return [dict(row) for row in rows]
finally:
conn.close()
def main() -> int:
parser = argparse.ArgumentParser(description="Initialize sort local SQLite storage.")
parser.add_argument("--st-root", required=True, help="Absolute path to the _St directory")
parser.add_argument("--db", help="SQLite path. Defaults to _St/Shared/data/sort_local.sqlite3")
parser.add_argument("--no-create-storage-dirs", action="store_true")
args = parser.parse_args()
st_root = Path(args.st_root)
db_path = Path(args.db) if args.db else default_database_path(st_root)
try:
result = initialize_database(
db_path,
st_root,
create_storage_dirs=not args.no_create_storage_dirs,
)
except (OSError, sqlite3.Error, ValueError) as exc:
print(json.dumps({"overall": "FAIL", "error": str(exc)}, ensure_ascii=False))
return 1
print(json.dumps({"overall": "OK", **result}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())