Refactor code structure for improved readability and maintainability

LR4/text_similarity_analysis.py

@@ -0,0 +1,220 @@
+from collections import Counter, defaultdict
+from pathlib import Path
+import math
+import re
+
+import matplotlib.pyplot as plt
+import numpy as np
+from matplotlib import rcParams
+
+rcParams["font.sans-serif"] = ["DejaVu Sans"]
+rcParams["axes.unicode_minus"] = False
+
+try:
+    import spacy
+
+    NLP = spacy.blank("ru")
+except Exception:
+    NLP = None
+
+
+def tok(text):
+    text = text.lower()
+    if NLP is not None:
+        return [t.text for t in NLP(text) if t.text.strip() and t.is_alpha]
+    return re.findall(r"\w+", text, flags=re.UNICODE)
+
+
+def make_windows(words, k):
+    return [(i, " ".join(words[i : i + k]), set(words[i : i + k])) for i in range(max(0, len(words) - k + 1))]
+
+
+def lev(a, b):
+    if a == b:
+        return 1.0
+    if not a or not b:
+        return 0.0
+    if len(a) < len(b):
+        a, b = b, a
+    prev = list(range(len(b) + 1))
+    for i, c1 in enumerate(a, 1):
+        cur = [i]
+        for j, c2 in enumerate(b, 1):
+            cur.append(min(cur[-1] + 1, prev[j] + 1, prev[j - 1] + (c1 != c2)))
+        prev = cur
+    return 1 - prev[-1] / max(len(a), len(b))
+
+
+def jaro(a, b):
+    if a == b:
+        return 1.0
+    if not a or not b:
+        return 0.0
+    la, lb = len(a), len(b)
+    d = max(0, max(la, lb) // 2 - 1)
+    ma, mb, m = [0] * la, [0] * lb, 0
+    for i in range(la):
+        for j in range(max(0, i - d), min(i + d + 1, lb)):
+            if not mb[j] and a[i] == b[j]:
+                ma[i] = mb[j] = 1
+                m += 1
+                break
+    if m == 0:
+        return 0.0
+    t, j = 0, 0
+    for i in range(la):
+        if not ma[i]:
+            continue
+        while not mb[j]:
+            j += 1
+        t += a[i] != b[j]
+        j += 1
+    return (m / la + m / lb + (m - t / 2) / m) / 3
+
+
+def winkler(a, b, p=0.1, bt=0.7):
+    dj = jaro(a, b)
+    if dj <= bt:
+        return dj
+    pref = 0
+    for x, y in zip(a[:4], b[:4]):
+        if x != y:
+            break
+        pref += 1
+    return dj + pref * p * (1 - dj)
+
+
+def jaccard(a, b, n=2):
+    def ng(s):
+        if len(s) < n:
+            return {s} if s else set()
+        return {s[i : i + n] for i in range(len(s) - n + 1)}
+
+    x, y = ng(a), ng(b)
+    if not x and not y:
+        return 1.0
+    if not x or not y:
+        return 0.0
+    return len(x & y) / len(x | y)
+
+
+def best_thr(vals, method):
+    if vals.size == 0:
+        return 1.0
+    grid = np.arange(0.65, 1.001, 0.01) if method in {"jaro", "jaro_winkler"} else (
+        np.arange(0.55, 1.001, 0.01) if method == "levenshtein" else np.arange(0.35, 1.001, 0.01)
+    )
+    best_t, best_q = float(grid[0]), -1.0
+    for t in grid:
+        s = vals[vals >= t]
+        if s.size == 0:
+            continue
+        q = 0.65 * float(np.mean(s)) + 0.35 * (s.size / vals.size)
+        if q > best_q:
+            best_t, best_q = float(t), q
+    return best_t
+
+
+def score_pairs(w1, w2, sim, min_overlap):
+    idx = defaultdict(list)
+    for j, (_, _, terms) in enumerate(w2):
+        for term in terms:
+            idx[term].append(j)
+
+    out = []
+    for i, p1, t1 in w1:
+        c = Counter()
+        for term in t1:
+            for j in idx.get(term, []):
+                c[j] += 1
+        for j, ov in c.items():
+            if ov >= min_overlap:
+                out.append((p1, sim(p1, w2[j][1])))
+    return out
+
+
+def analyze(words1, words2, ws=(5, 10, 20, 30), n=2, p=0.1, bt=0.7):
+    methods = {
+        "levenshtein": lev,
+        "jaro": jaro,
+        "jaro_winkler": lambda a, b: winkler(a, b, p, bt),
+        "jaccard": lambda a, b: jaccard(a, b, n),
+    }
+    stats = defaultdict(dict)
+
+    print("=" * 64)
+    print("АНАЛИЗ СХОЖЕСТИ ТЕКСТОВ")
+    print("=" * 64)
+    print(f"Текст 1: {len(words1)} слов")
+    print(f"Текст 2: {len(words2)} слов")
+    print(f"Токенизация: {'SpaCy' if NLP is not None else 'regex fallback'}")
+
+    for name, sim in methods.items():
+        print("\n" + "-" * 64)
+        print(name.upper())
+        print("-" * 64)
+        for k in ws:
+            w1, w2 = make_windows(words1, k), make_windows(words2, k)
+            scored = score_pairs(w1, w2, sim, max(1, math.ceil(k * 0.35)))
+            vals = np.array([x[1] for x in scored], dtype=float)
+            thr = best_thr(vals, name)
+            matched = [x for x in scored if x[1] >= thr]
+            total = sum(len(x[0]) for x in matched)
+            avg = float(np.mean([x[1] for x in matched])) if matched else 0.0
+            stats[name][k] = {
+                "threshold": thr,
+                "matches_count": len(matched),
+                "total_length": int(total),
+                "avg_similarity": avg,
+            }
+            print(f"окно={k:>2} | порог={thr:.2f} | совпадений={len(matched):>4} | длина={int(total):>5} | ср={avg:.3f}")
+    return stats
+
+
+def plot(stats, ws, out):
+    methods = list(stats)
+    fig, ax = plt.subplots(2, 2, figsize=(14, 10))
+    fig.suptitle("Схожесть текстов", fontsize=14)
+    items = [
+        ("matches_count", "Количество совпадений", "o", None),
+        ("total_length", "Суммарная длина", "s", None),
+        ("avg_similarity", "Средняя схожесть", "^", (0, 1)),
+        ("threshold", "Оптимальные пороги", "D", (0, 1)),
+    ]
+    for a, (key, title, marker, ylim) in zip(ax.flat, items):
+        for m in methods:
+            a.plot(ws, [stats[m][k][key] for k in ws], marker=marker, linewidth=2, label=m)
+        a.set_title(title)
+        a.set_xlabel("Размер окна (слов)")
+        if ylim:
+            a.set_ylim(*ylim)
+        a.grid(alpha=0.3)
+        a.legend()
+    plt.tight_layout()
+    img = out / "similarity_analysis.png"
+    plt.savefig(img, dpi=250, bbox_inches="tight")
+    print(f"\nСохранен график: {img}")
+
+
+def main():
+    base = Path(__file__).resolve().parent
+    words1 = tok((base / "hameleon_short.txt").read_text(encoding="utf-8"))
+    words2 = tok((base / "hameleon.txt").read_text(encoding="utf-8"))
+    ws = [5, 10, 20, 30]
+
+    stats = analyze(words1, words2, ws=ws, n=2, p=0.1, bt=0.7)
+
+    print("\n" + "=" * 64)
+    print("ИТОГОВАЯ СВОДКА")
+    print("=" * 64)
+    for m in stats:
+        cnt = sum(stats[m][k]["matches_count"] for k in ws)
+        ln = sum(stats[m][k]["total_length"] for k in ws)
+        print(f"{m:>13}: совпадений={cnt:>4}, суммарная длина={ln:>5}")
+
+    plot(stats, ws, base)
+    print("\nАНАЛИЗ ЗАВЕРШЕН")
+
+
+if __name__ == "__main__":
+    main()