bingshuo/guanghu-ice-heart
Watch
1
0
Fork
You've already forked guanghu-ice-heart
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

fix(runtime): archive duplicate daily life candidates

Browse source
This commit is contained in:
冰朔 2026-09-13 11:05:48 +08:00
commit a2f23b49ff
5 changed files with 349 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

45
server-tools/tcs-mother-body/archive_duplicate_daily_reviews.py
View file

@ -56,7 +56,9 @@ def main():
with lock_path.open("w") as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
selected = []
selected_inbox = []
before = collections.Counter()
inbox_before = 0
nonreview = 0
personas = root / "personas"
for persona in sorted(personas.iterdir()):
@ -86,10 +88,45 @@ def main():
if item is not keeper:
item["kept_job_id"] = keeper["job_id"]
selected.append(item)
life_line = persona / "life-line"
selected_candidate_ids = set()
try:
current = load(life_line / "CURRENT.json")
selected_candidate_ids = {item.get("selected_candidate_id") for item in current.get("blocks", []) if item.get("selected_candidate_id")}
except (OSError, ValueError):
pass
inbox = life_line / "inbox"
inbox_groups = collections.defaultdict(list)
if inbox.is_dir():
for day in sorted(inbox.iterdir()):
if not day.is_dir():
continue
for candidate in day.iterdir():
if not candidate.is_file() or not candidate.name.endswith(".json"):
continue
try:
value = load(candidate)
except (OSError, ValueError):
continue
event_id = str(value.get("source_event_id", ""))
if value.get("source_type") != "ONLINE_PERSONA_EVENT" or not event_id.startswith("DAILY-REVIEW-"):
continue
inbox_before += 1
inbox_groups[(day.name, event_id)].append({"persona_id": persona.name, "event_id": event_id,
"candidate_id": candidate.stem, "day": day.name, "path": candidate,
"selected_by_life_line": candidate.stem in selected_candidate_ids})
for group in inbox_groups.values():
keeper = min(group, key=lambda item: (not item["selected_by_life_line"], item["candidate_id"]))
for item in group:
if item is not keeper:
item["kept_candidate_id"] = keeper["candidate_id"]
selected_inbox.append(item)
summary = {"schema": "guanghu.daily-review-duplicate-archive/v1", "batch": args.batch,
"root": str(root), "mode": "APPLY" if args.apply else "DRY_RUN",
"review_jobs_before": sum(before.values()), "review_state_counts_before": dict(before),
"duplicate_jobs_selected": len(selected), "active_review_jobs_after": sum(before.values()) - len(selected),
"review_inbox_candidates_before": inbox_before, "duplicate_inbox_candidates_selected": len(selected_inbox),
"active_review_inbox_candidates_after": inbox_before - len(selected_inbox),
"nonreview_jobs_untouched": nonreview, "lossless_move": True}
if not args.apply:
print(json.dumps(summary, ensure_ascii=False, indent=2))
@ -109,6 +146,14 @@ def main():
record = {**item, "source": str(source), "archive": str(destination), "file_sha256": files}
os.rename(source, destination)
output.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
for item in selected_inbox:
source = item.pop("path")
destination = archive_root / "life-inbox" / item["persona_id"] / item["day"] / source.name
destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
record = {**item, "kind": "DAILY_REVIEW_LIFE_TIME_CANDIDATE", "source": str(source),
"archive": str(destination), "file_sha256": sha256(source)}
os.rename(source, destination)
output.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
output.flush()
os.fsync(output.fileno())
summary["manifest"] = str(manifest)