131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
"""将 rules / accounts 中卡片回复的 localhost URL 刷新为 KEFU_PUBLIC_BASE_URL。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
load_dotenv(ROOT / ".env")
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from link_cards import absolute_media_url, public_base_url # noqa: E402
|
|
|
|
|
|
def _card_lookup(conn: sqlite3.Connection) -> dict[int, sqlite3.Row]:
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute(
|
|
"SELECT id, slug, image_path FROM link_card_pages"
|
|
).fetchall()
|
|
return {int(r["id"]): r for r in rows}
|
|
|
|
|
|
def _refresh_card_item(item: dict, cards: dict[int, sqlite3.Row], base: str) -> bool:
|
|
if item.get("type") != "card":
|
|
return False
|
|
changed = False
|
|
card_id = item.get("card_id")
|
|
slug = None
|
|
image_path = (item.get("image_path") or "").strip()
|
|
if card_id is not None:
|
|
row = cards.get(int(card_id))
|
|
if row:
|
|
slug = row["slug"]
|
|
if not image_path:
|
|
image_path = (row["image_path"] or "").strip()
|
|
if slug and base:
|
|
new_url = f"{base}/p/{slug}"
|
|
if item.get("url") != new_url:
|
|
item["url"] = new_url
|
|
changed = True
|
|
if image_path:
|
|
new_cover = absolute_media_url(image_path)
|
|
if new_cover and item.get("cover_url") != new_cover:
|
|
item["cover_url"] = new_cover
|
|
changed = True
|
|
if item.get("image_path") != image_path:
|
|
item["image_path"] = image_path
|
|
changed = True
|
|
return changed
|
|
|
|
|
|
def _refresh_reply_content(raw: str | None, cards: dict[int, sqlite3.Row], base: str) -> tuple[str | None, bool]:
|
|
if not raw or not raw.strip():
|
|
return raw, False
|
|
text = raw.strip()
|
|
if not text.startswith("{"):
|
|
return raw, False
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return raw, False
|
|
|
|
changed = False
|
|
if isinstance(data, dict) and data.get("type") == "card":
|
|
changed = _refresh_card_item(data, cards, base)
|
|
return (json.dumps(data, ensure_ascii=False), changed) if changed else (raw, False)
|
|
|
|
messages = data.get("messages") if isinstance(data, dict) else None
|
|
if not isinstance(messages, list):
|
|
return raw, False
|
|
for item in messages:
|
|
if isinstance(item, dict):
|
|
changed = _refresh_card_item(item, cards, base) or changed
|
|
if not changed:
|
|
return raw, False
|
|
return json.dumps(data, ensure_ascii=False), True
|
|
|
|
|
|
def main() -> None:
|
|
base = public_base_url()
|
|
if not base:
|
|
print("错误: 请先在 .env 设置 KEFU_PUBLIC_BASE_URL=https://你的公网域名")
|
|
sys.exit(1)
|
|
print(f"公网基址: {base}")
|
|
|
|
db_path = Path(__file__).resolve().parent / "kefu.db"
|
|
conn = sqlite3.connect(db_path)
|
|
cards = _card_lookup(conn)
|
|
|
|
rule_updates = 0
|
|
for row in conn.execute("SELECT id, keyword, reply_content FROM rules"):
|
|
rid, keyword, content = row
|
|
new_content, changed = _refresh_reply_content(content, cards, base)
|
|
if changed:
|
|
conn.execute(
|
|
"UPDATE rules SET reply_content = ? WHERE id = ?",
|
|
(new_content, rid),
|
|
)
|
|
rule_updates += 1
|
|
print(f"已更新规则 id={rid} keyword={keyword!r}")
|
|
print(f" -> {new_content[:200]}...")
|
|
|
|
account_updates = 0
|
|
for row in conn.execute(
|
|
"SELECT id, username, follow_welcome_content FROM accounts WHERE follow_welcome_content IS NOT NULL"
|
|
):
|
|
aid, username, content = row
|
|
new_content, changed = _refresh_reply_content(content, cards, base)
|
|
if changed:
|
|
conn.execute(
|
|
"UPDATE accounts SET follow_welcome_content = ? WHERE id = ?",
|
|
(new_content, aid),
|
|
)
|
|
account_updates += 1
|
|
print(f"已更新账号 id={aid} username={username!r} 关注欢迎语")
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"完成: 规则 {rule_updates} 条, 账号欢迎语 {account_updates} 条")
|
|
if rule_updates or account_updates:
|
|
print("请重启后端并重新启动账号托管后测试卡片回复。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|