316 lines
11 KiB
Python
316 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
xiaoguo_quick_scan.py — 小果高频扫描（每15分钟）
|
||
|
||
限制：每次最多处理 3 篇最新的不达标文章，避免积压。
|
||
零成本运行。
|
||
"""
|
||
|
||
import subprocess, sys, json, os, re, random
|
||
from pathlib import Path
|
||
from datetime import datetime, timezone
|
||
|
||
RAW = Path("/home/hmo/Obsidian/raw/articles")
|
||
WIKI = Path("/home/hmo/Obsidian/wiki")
|
||
SYNTHESIS = WIKI / "synthesis"
|
||
ENTITIES = WIKI / "entities"
|
||
CATEGORIES = WIKI / "categories"
|
||
LINKS_FILE = Path("/home/hmo/Obsidian/wechat-article-links.md")
|
||
STATE_FILE = Path("/tmp/xiaoguo_scan_state.json")
|
||
LESSONS_FILE = Path("/home/hmo/Obsidian/wiki/structured_lessons.md")
|
||
MAX_PER_RUN = 3
|
||
|
||
def call_xiaoguo(prompt: str, max_tokens=200, max_retries=3) -> str:
|
||
"""调小果 Ollama，带排队锁和重试。
|
||
|
||
锁文件 /tmp/xiaoguo_busy.lock 防止多源并发冲突。
|
||
每次请求最多等 30 秒获取锁，拿到后 120 秒超时。
|
||
"""
|
||
import time, os, fcntl
|
||
|
||
lock_path = "/tmp/xiaoguo_busy.lock"
|
||
|
||
for attempt in range(max_retries):
|
||
# 尝试获取锁
|
||
lock_acquired = False
|
||
try:
|
||
with open(lock_path, "w") as lf:
|
||
fcntl.flock(lf, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
lock_acquired = True
|
||
# 释放锁，让它随 with 块结束自动释放
|
||
except (IOError, OSError):
|
||
lock_acquired = False
|
||
|
||
if not lock_acquired:
|
||
if attempt < max_retries - 1:
|
||
time.sleep(5 * (attempt + 1)) # 5s, 10s, 15s 退避
|
||
continue
|
||
return ""
|
||
|
||
try:
|
||
# 真正调小果
|
||
payload = {
|
||
"model": "Qwen3.6-27B-MTPLX-Optimized-Speed",
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"max_tokens": max_tokens,
|
||
"temperature": 0.1,
|
||
}
|
||
result = subprocess.run(
|
||
["curl", "-s", "--max-time", "120", "-X", "POST",
|
||
"http://127.0.0.1:18003/v1/chat/completions",
|
||
"-H", "Content-Type: application/json",
|
||
"-d", json.dumps(payload)],
|
||
capture_output=True, text=True, timeout=130
|
||
)
|
||
data = json.loads(result.stdout)
|
||
return data["choices"][0]["message"]["content"].strip()
|
||
except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, Exception) as e:
|
||
if attempt < max_retries - 1:
|
||
time.sleep(3 * (attempt + 1))
|
||
continue
|
||
return ""
|
||
finally:
|
||
try:
|
||
os.remove(lock_path)
|
||
except:
|
||
pass
|
||
|
||
return ""
|
||
|
||
def get_processed_ids():
|
||
if STATE_FILE.exists():
|
||
return set(json.loads(STATE_FILE.read_text()).get("processed", []))
|
||
return set()
|
||
|
||
def save_processed_ids(ids):
|
||
STATE_FILE.write_text(json.dumps({"processed": list(ids), "updated": datetime.now().isoformat()}))
|
||
|
||
def check_frontmatter(content: str) -> list:
|
||
missing = []
|
||
if not content.startswith("---"):
|
||
return ["frontmatter"]
|
||
end = content.find("---", 3)
|
||
fm = content[3:end] if end > 0 else ""
|
||
for field in ["original_wechat_url", "wechat_status"]:
|
||
if not re.search(rf"^{field}:", fm, re.MULTILINE):
|
||
missing.append(field)
|
||
return missing
|
||
|
||
def batch_fix(articles: list) -> int:
|
||
"""批量补 frontmatter。"""
|
||
if not articles:
|
||
return 0
|
||
|
||
batch_prompt = ""
|
||
for fpath, missing in articles:
|
||
content = fpath.read_text(encoding="utf-8", errors="replace")
|
||
batch_prompt += f"--- {fpath.name} ---\n内容前300字: {content[:300]}\n缺失字段: {missing}\n\n"
|
||
|
||
prompt = f"""以下每篇文章缺 frontmatter 字段。对每篇输出一行修复建议，格式：「文件名: 字段名=值」。
|
||
只输出有把握的，不知道的写「字段名=unknown」。
|
||
|
||
{batch_prompt}"""
|
||
|
||
payload = {
|
||
"model": "Qwen3.6-27B-MTPLX-Optimized-Speed",
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"max_tokens": 500,
|
||
"temperature": 0.1,
|
||
}
|
||
|
||
try:
|
||
result = subprocess.run(
|
||
["curl", "-s", "--max-time", "120", "-X", "POST",
|
||
"http://127.0.0.1:18003/v1/chat/completions",
|
||
"-H", "Content-Type: application/json",
|
||
"-d", json.dumps(payload)],
|
||
capture_output=True, text=True, timeout=130
|
||
)
|
||
data = json.loads(result.stdout)
|
||
output = data["choices"][0]["message"]["content"].strip()
|
||
except Exception as e:
|
||
print(f" 小果调用失败: {e}")
|
||
return 0
|
||
|
||
# 解析输出并更新文件
|
||
fixed = 0
|
||
for fpath, missing in articles:
|
||
content_old = fpath.read_text(encoding="utf-8", errors="replace")
|
||
lines = content_old.split("\n")
|
||
fm_end = 0
|
||
for i, line in enumerate(lines):
|
||
if i > 0 and line.strip() == "---":
|
||
fm_end = i
|
||
break
|
||
|
||
updated = False
|
||
for line in output.split("\n"):
|
||
if fpath.name in line and "=" in line:
|
||
parts = line.split("=", 1)
|
||
if len(parts) == 2:
|
||
key, val = parts[0].strip().split(":")[-1].strip(), parts[1].strip()
|
||
if val and val != "unknown":
|
||
# Check if already exists in frontmatter
|
||
exists = any(l.startswith(f"{key}:") for l in lines[1:fm_end])
|
||
if not exists:
|
||
lines.insert(fm_end, f"{key}: {val}")
|
||
fm_end += 1
|
||
updated = True
|
||
|
||
if updated:
|
||
fpath.write_text("\n".join(lines), encoding="utf-8")
|
||
fixed += 1
|
||
|
||
return fixed
|
||
|
||
def find_connection_candidates() -> list:
|
||
"""找一篇孤立/弱连接的实体，建议它应该连到哪个分类。"""
|
||
if not ENTITIES.exists():
|
||
return []
|
||
|
||
orphans = []
|
||
all_wiki_content = ""
|
||
for f in WIKI.rglob("*.md"):
|
||
all_wiki_content += f.read_text(encoding="utf-8", errors="replace")[:500] + "\n"
|
||
|
||
for f in sorted(ENTITIES.glob("*.md"), key=lambda x: -x.stat().st_mtime)[:5]:
|
||
name = f.stem
|
||
# 检查被引用次数
|
||
refs = all_wiki_content.count(f"[[{name}]]")
|
||
if refs <= 1: # 只被自己引用或没人引
|
||
content = f.read_text(encoding="utf-8", errors="replace")[:300]
|
||
orphans.append((f, name, content, refs))
|
||
|
||
return orphans
|
||
|
||
def quality_check_one_synthesis() -> dict:
|
||
"""抽查一篇合成报告，检查基本质量。"""
|
||
if not SYNTHESIS.exists():
|
||
return {}
|
||
|
||
reports = list(SYNTHESIS.glob("*.md"))
|
||
if not reports:
|
||
return {}
|
||
|
||
# 随机选一篇
|
||
f = random.choice(reports)
|
||
content = f.read_text(encoding="utf-8", errors="replace")
|
||
|
||
issues = []
|
||
|
||
# 检查 updated 日期
|
||
updated_m = re.search(r"^updated:\s*(\d{4}-\d{2}-\d{2})", content, re.MULTILINE)
|
||
if updated_m:
|
||
updated = datetime.strptime(updated_m.group(1), "%Y-%m-%d")
|
||
days_old = (datetime.now() - updated).days
|
||
if days_old > 30:
|
||
issues.append(f"已 {days_old} 天未更新")
|
||
|
||
# 检查 frontmatter
|
||
if not content.startswith("---"):
|
||
issues.append("缺 frontmatter")
|
||
|
||
# 检查长度
|
||
if len(content) < 500:
|
||
issues.append("内容过少")
|
||
|
||
# 检查 cross-links
|
||
if "cross:" not in content[:500]:
|
||
issues.append("缺 cross-links")
|
||
|
||
return {"file": f.name, "issues": issues, "days_old": days_old if updated_m else 0}
|
||
|
||
def main():
|
||
print(f"小果高频扫描 — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
||
print(f" 任务轮次: {datetime.now().timestamp() % 3:.0f}")
|
||
|
||
# 轮换任务，每次做一类
|
||
task_round = int(datetime.now().timestamp() / 900) % 3 # 每900秒(15分)轮换
|
||
|
||
if task_round == 0:
|
||
# Round 0: 补 frontmatter + links（已有逻辑）
|
||
processed = get_processed_ids()
|
||
needs_fix = []
|
||
all_files = sorted(RAW.glob("*.md"), key=lambda f: -f.stat().st_mtime)
|
||
for f in all_files:
|
||
if f.stem in processed:
|
||
continue
|
||
content = f.read_text(encoding="utf-8", errors="replace")
|
||
missing = check_frontmatter(content)
|
||
if missing:
|
||
needs_fix.append((f, missing))
|
||
if len(needs_fix) >= MAX_PER_RUN:
|
||
break
|
||
|
||
if needs_fix:
|
||
print(f" 补 frontmatter: {len(needs_fix)} 篇")
|
||
fixed = batch_fix(needs_fix)
|
||
new_processed = set(processed)
|
||
for f, _ in needs_fix:
|
||
new_processed.add(f.stem)
|
||
save_processed_ids(new_processed)
|
||
|
||
# 补 links
|
||
linked = 0
|
||
for f, _ in needs_fix:
|
||
if LINKS_FILE.exists():
|
||
lc = LINKS_FILE.read_text(encoding="utf-8")
|
||
if f.stem not in lc:
|
||
with open(LINKS_FILE, "a", encoding="utf-8") as lf:
|
||
lf.write(f"{datetime.now().strftime('%Y-%m-%d')} | {f.stem} | （小果补充） | （待补URL）\n")
|
||
linked += 1
|
||
print(f" 补 links: {linked}")
|
||
else:
|
||
print(" 无待补文章，跳至连接挖掘")
|
||
# frontmatter 补完了就做连接挖掘
|
||
orphans = find_connection_candidates()
|
||
if orphans:
|
||
f, name, content, refs = orphans[0]
|
||
print(f" 孤立实体: {name} (引用{refs}次)")
|
||
|
||
elif task_round == 1:
|
||
# Round 1: 连接挖掘 + 跨链接发现
|
||
orphans = find_connection_candidates()
|
||
if orphans:
|
||
f, name, content, refs = orphans[0]
|
||
prompt = f"""实体名称: {name}
|
||
实体内容: {content[:300]}
|
||
引用次数: {refs}
|
||
|
||
从 wiki 分类中找出最应该链接到这个实体的分类。
|
||
可用分类: {', '.join(sorted([c.stem for c in CATEGORIES.glob('*.md')])[:20])}
|
||
|
||
只回答分类名，不解释。"""
|
||
|
||
result = call_xiaoguo(prompt)
|
||
if result:
|
||
print(f" 🔗 建议 {name} → {result.strip()}")
|
||
else:
|
||
print(f" 无连接建议")
|
||
else:
|
||
print(" 无孤立实体")
|
||
|
||
# 质量检查
|
||
qc = quality_check_one_synthesis()
|
||
if qc and qc.get("issues"):
|
||
print(f" 质量: {qc['file']} — {', '.join(qc['issues'])}")
|
||
elif qc:
|
||
print(f" 质量: {qc['file']} — OK")
|
||
|
||
elif task_round == 2:
|
||
# Round 2: 教训学习 — 从 self_remediation.md 读取未处理项
|
||
remediation = Path("/home/hmo/Obsidian/self_remediation.md")
|
||
if remediation.exists():
|
||
content = remediation.read_text(encoding="utf-8")
|
||
todo_items = re.findall(r"## (.+?)\n.*?问题: (.+?)\n.*?状态: 待修正", content, re.DOTALL)
|
||
if todo_items:
|
||
slug, issue = todo_items[0]
|
||
print(f" 📋 待修正: {slug} — {issue.strip()[:100]}")
|
||
else:
|
||
print(f" 无待修正项")
|
||
else:
|
||
print(f" 无待修正项")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|