#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
法官橫向網絡
============
從 40_案件歷程/ 的串接結果，建立「承審過後來被聲請釋憲之案件」的法官網絡。
python _pipeline\\法官網絡.py
產出於 40_案件歷程/：
法官統計.csv 每位法官的承審件數、涉及釋憲案、法院、審級、共事人數
合議庭共事.csv 兩兩共事次數與共同案件
法官網絡.html 離線互動式網絡圖（力導向，無外部相依，雙擊即開）
口徑說明：
* 節點為法官，邊為「同一案號同時列名」，即合議庭共事關係。
* 「承審件數」以案號計，不重複計算同一案號。
* 資料僅涵蓋有串接成功的釋憲原因案件，不是法官的全部辦案紀錄，
因此排名意義是「在釋憲原因案件中出現的頻率」，不可解讀為個人辦案量。
"""
import csv, json, collections, sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
D = ROOT / "40_案件歷程"
def court_kind(c):
if c.startswith("最高行政"):
return "最高行政法院"
if c.startswith("最高"):
return "最高法院"
if "高等行政" in c:
return "高等行政法院"
if "高等法院" in c:
return "高等法院"
if "地方法院" in c:
return "地方法院"
return "其他"
def main():
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
src = D / "法官參與.csv"
if not src.exists():
print("找不到", src, "\n請先執行 _pipeline\\串接案件歷程.py")
sys.exit(1)
rows = list(csv.DictReader(open(src, encoding="utf-8-sig")))
rows = [r for r in rows if r.get("法官", "").strip()]
print(f"讀入承審紀錄 {len(rows)} 列")
# 案號 → 該案的法官集合、法院、審級、涉及之釋憲案
bycase = collections.defaultdict(lambda: {"法官": set(), "法院": "", "審級": "", "釋憲": set()})
for r in rows:
c = bycase[r["案號"]]
c["法官"].add(r["法官"].strip())
c["法院"] = r["法院"]
c["審級"] = r["審級"]
c["釋憲"].add(r["釋憲案"])
# 法官統計
stat = collections.defaultdict(lambda: {"案號": set(), "釋憲": set(), "法院": collections.Counter(),
"審級": collections.Counter(), "夥伴": collections.Counter()})
for case, c in bycase.items():
js = sorted(c["法官"])
for j in js:
s = stat[j]
s["案號"].add(case)
s["釋憲"] |= c["釋憲"]
s["法院"][c["法院"]] += 1
s["審級"][c["審級"]] += 1
for a in js:
for b in js:
if a != b:
stat[a]["夥伴"][b] += 1
with open(D / "法官統計.csv", "w", encoding="utf-8-sig", newline="") as fh:
w = csv.writer(fh)
w.writerow(["法官", "承審件數", "涉及釋憲案數", "涉及釋憲案", "主要法院", "所有法院",
"審級分布", "共事法官人數", "最常共事者"])
for j, s in sorted(stat.items(), key=lambda x: (-len(x[1]["案號"]), x[0])):
top = "、".join(f"{k}({v})" for k, v in s["夥伴"].most_common(3))
w.writerow([j, len(s["案號"]), len(s["釋憲"]), "、".join(sorted(s["釋憲"])),
s["法院"].most_common(1)[0][0] if s["法院"] else "",
"、".join(sorted(s["法院"])),
"、".join(f"{k}:{v}" for k, v in s["審級"].most_common()),
len(s["夥伴"]), top])
# 共事邊
pair = collections.Counter()
paircase = collections.defaultdict(set)
for case, c in bycase.items():
js = sorted(c["法官"])
for i in range(len(js)):
for k in range(i + 1, len(js)):
pair[(js[i], js[k])] += 1
paircase[(js[i], js[k])].add(case)
with open(D / "合議庭共事.csv", "w", encoding="utf-8-sig", newline="") as fh:
w = csv.writer(fh)
w.writerow(["法官A", "法官B", "共事案件數", "共同案件"])
for (a, b), n in sorted(pair.items(), key=lambda x: -x[1]):
w.writerow([a, b, n, "；".join(sorted(paircase[(a, b)]))])
print(f"法官 {len(stat)} 位；共事關係 {len(pair)} 組；案件 {len(bycase)} 件")
# ---------------- 網絡圖 ----------------
nodes = []
idx = {}
for j, s in sorted(stat.items(), key=lambda x: (-len(x[1]["案號"]), x[0])):
idx[j] = len(nodes)
main_court = s["法院"].most_common(1)[0][0] if s["法院"] else ""
nodes.append({"n": j, "v": len(s["案號"]), "r": len(s["釋憲"]),
"c": court_kind(main_court), "ct": main_court,
"p": len(s["夥伴"]),
"z": "、".join(sorted(s["釋憲"]))[:120]})
links = [{"s": idx[a], "t": idx[b], "w": n} for (a, b), n in pair.items()]
data = json.dumps({"nodes": nodes, "links": links}, ensure_ascii=False).replace("<", "\\u003c")
html = HTML.replace("__DATA__", data).replace("__NCASE__", str(len(bycase)))
(D / "法官網絡.html").write_text(html, encoding="utf-8")
print("網絡圖：", D / "法官網絡.html")
HTML = """
法官橫向網絡
法官橫向網絡
承審過後來被聲請釋憲之案件的法官與其合議庭共事關係 共 __NCASE__ 件原因案件
節點大小為承審件數，連線粗細為共事次數。此處統計僅涵蓋串接成功之釋憲原因案件，不代表法官全部辦案量。
"""
if __name__ == "__main__":
main()