diff --git "a/optimizer.py" "b/optimizer.py"
deleted file mode 100644--- "a/optimizer.py"
+++ /dev/null
@@ -1,3089 +0,0 @@
-import json
-import difflib
-import html as html_lib
-import re
-from itertools import combinations
-from typing import Any, Dict, List, Optional, Tuple
-
-import requests
-
-import logic
-import nlp_processor
-import semantic_graph
-
-
-STOP_WORDS = {
- "en": {"a", "an", "and", "or", "the", "to", "of", "for", "in", "on", "at", "by", "with", "from", "as", "is", "are", "be", "was", "were"},
- "ru": {"и", "или", "в", "во", "на", "по", "с", "со", "к", "ко", "для", "из", "за", "что", "это", "как", "а", "но", "у", "о", "от"},
- "de": {"und", "oder", "der", "die", "das", "zu", "von", "mit", "fur", "in", "auf", "ist", "sind"},
- "es": {"y", "o", "el", "la", "los", "las", "de", "del", "en", "con", "para", "por", "es", "son"},
- "it": {"e", "o", "il", "lo", "la", "i", "gli", "le", "di", "del", "in", "con", "per", "da", "e", "sono"},
- "pl": {"i", "oraz", "lub", "w", "na", "z", "ze", "do", "od", "po", "dla", "to", "jest", "sa"},
- "pt": {"e", "ou", "o", "a", "os", "as", "de", "do", "da", "em", "no", "na", "com", "para", "por", "e", "sao"},
-}
-
-BERT_TARGET_THRESHOLD = 0.7
-BERT_GOAL_DELTA_MIN = 0.005
-TITLE_TARGET_THRESHOLD = 0.65
-SEMANTIC_GAP_TOLERANCE_PCT = 0.15
-SEMANTIC_GAP_MIN_ABS = 3.0
-STAGE_ORDER = ["bert", "bm25", "ngram", "semantic", "title"]
-NGRAM_ATTEMPTS_PER_TERM = 3
-
-
-def _normalize_stage_name(v: Any) -> str:
- s = str(v or "").strip().lower()
- return s if s in STAGE_ORDER else ""
-
-
-def _goal_label_canonical(goal: Dict[str, Any]) -> str:
- t = str(goal.get("type", "") or "").strip().lower()
- if t == "bm25":
- term = str(goal.get("bm25_word", "") or "").strip().lower()
- if term:
- return term
- label = str(goal.get("label", "") or "").strip().lower()
- if label.startswith("reduce spam:"):
- return label.replace("reduce spam:", "", 1).strip()
- return label
- return str(goal.get("label", "") or "").strip().lower()
-
-
-def _build_custom_goal(stage: str, value: str, language: str) -> Optional[Dict[str, Any]]:
- raw = str(value or "").strip()
- if not raw:
- return None
- if stage == "bert":
- return {
- "type": "bert",
- "label": raw,
- "focus_terms": _filter_stopwords(_tokenize(raw), language)[:6],
- "avoid_terms": [],
- "bert_phrase_score": 0.0,
- "bert_target": float(BERT_TARGET_THRESHOLD),
- }
- if stage == "bm25":
- return {
- "type": "bm25",
- "label": f"reduce spam: {raw}",
- "focus_terms": [],
- "avoid_terms": [raw],
- "bm25_count": 2,
- "bm25_word": raw,
- }
- if stage == "ngram":
- return {
- "type": "ngram",
- "label": raw,
- "focus_terms": [raw],
- "avoid_terms": [],
- "ngram_target_count": 0.0,
- "ngram_comp_avg": 1.0,
- "ngram_tolerance_pct": 0.5,
- "ngram_lower_bound": 0.5,
- "ngram_upper_bound": 1.5,
- "ngram_direction": "increase",
- "ngram_rank_index": 0,
- "ngram_candidates_total": 1,
- }
- if stage == "semantic":
- return {
- "type": "semantic",
- "label": raw,
- "focus_terms": [raw],
- "avoid_terms": [],
- "semantic_gap": float(SEMANTIC_GAP_MIN_ABS),
- }
- if stage == "title":
- return {
- "type": "title",
- "label": "title alignment",
- "focus_terms": _filter_stopwords(_tokenize(raw), language)[:8],
- "avoid_terms": [],
- "title_bert_score": 0.0,
- "title_target": float(TITLE_TARGET_THRESHOLD),
- }
- return None
-
-
-def _apply_stage_goal_override(
- goals: List[Dict[str, Any]],
- stage: str,
- language: str,
- stage_goal_overrides: Optional[Dict[str, Any]],
-) -> List[Dict[str, Any]]:
- ov_all = stage_goal_overrides or {}
- ov = ov_all.get(stage) if isinstance(ov_all, dict) else None
- if not isinstance(ov, dict):
- return goals
-
- mode = str(ov.get("mode", "auto") or "auto").strip().lower()
- if mode not in {"auto", "manual", "mixed"}:
- mode = "auto"
-
- selected_raw = ov.get("selected") or []
- custom_raw = ov.get("custom_add") or []
- selected_set = {
- str(x or "").strip().lower()
- for x in selected_raw
- if str(x or "").strip()
- }
-
- custom_goals: List[Dict[str, Any]] = []
- for item in custom_raw:
- g = _build_custom_goal(stage, str(item or ""), language)
- if g:
- custom_goals.append(g)
-
- if mode == "auto":
- out = list(goals)
- else:
- out = []
- for g in goals:
- if _goal_label_canonical(g) in selected_set:
- out.append(g)
-
- if mode in {"manual", "mixed"}:
- out.extend(custom_goals)
-
- # Deduplicate by canonical goal label to keep deterministic cursor behavior.
- dedup: Dict[str, Dict[str, Any]] = {}
- for g in out:
- key = _goal_label_canonical(g)
- if key and key not in dedup:
- dedup[key] = g
- return list(dedup.values())
-
-
-def _tokenize(text: str) -> List[str]:
- return [
- x
- for x in re.sub(r"[^\w\s-]+", " ", (text or "").lower(), flags=re.UNICODE).split()
- if len(x) >= 2
- ]
-
-
-def _tokenize_ngrams_strict(text: str) -> List[str]:
- """
- Tokenizer for strict n-gram matching.
- В отличие от _tokenize() здесь НЕ отбрасываем короткие токены (например, "a"),
- чтобы совпадать с теми же n-граммами, которые генерирует логика BM25.
- """
- return [
- x
- for x in re.sub(r"[^\w\s-]+", " ", (text or "").lower(), flags=re.UNICODE).split()
- if x
- ]
-
-
-def _count_term_ngrams_strict(text: str, term: str) -> int:
- """
- Strict count of exact token n-gram occurrences (overlapping allowed).
- Считает вхождения term как последовательности токенов.
- """
- text_tokens = _tokenize_ngrams_strict(text)
- term_tokens = _tokenize_ngrams_strict(term)
- m = len(term_tokens)
- if m == 0 or not text_tokens or len(text_tokens) < m:
- return 0
-
- cnt = 0
- # Sliding window over token sequence
- for i in range(0, len(text_tokens) - m + 1):
- if text_tokens[i : i + m] == term_tokens:
- cnt += 1
- return cnt
-
-
-def _filter_stopwords(tokens: List[str], language: str) -> List[str]:
- stop = STOP_WORDS.get(language, STOP_WORDS["en"])
- return [t for t in tokens if t not in stop]
-
-
-def _split_sentences(text: str) -> List[str]:
- text = (text or "").strip()
- if not text:
- return []
- parts = re.split(r"(?<=[\.\!\?])\s+", text)
- parts = [p.strip() for p in parts if p.strip()]
- if len(parts) <= 1:
- parts = [p.strip() for p in re.split(r"\n+", text) if p.strip()]
- return parts
-
-
-def _escape_html(v: Any) -> str:
- return html_lib.escape(str(v or ""), quote=True)
-
-
-def _diff_sentences_html(before_text: str, after_text: str) -> Tuple[str, List[Dict[str, str]]]:
- """
- sentence-level diff for UI highlighting.
- Возвращает:
- - html для отображения ТОЛЬКО after_text (optimized),
- - список блоков что было/что стало (для "что именно менять").
- """
- before_sents = _split_sentences(before_text)
- after_sents = _split_sentences(after_text)
-
- matcher = difflib.SequenceMatcher(None, before_sents, after_sents, autojunk=False)
- parts: List[str] = []
- changes: List[Dict[str, str]] = []
-
- for tag, i1, i2, j1, j2 in matcher.get_opcodes():
- if tag == "equal":
- for j in range(j1, j2):
- parts.append(_escape_html(after_sents[j]))
- elif tag == "replace":
- from_txt = " ".join(before_sents[i1:i2]).strip()
- to_txt = " ".join(after_sents[j1:j2]).strip()
- changes.append({"type": "replace", "from": from_txt, "to": to_txt})
- for j in range(j1, j2):
- parts.append(
- f'{_escape_html(after_sents[j])}'
- )
- elif tag == "insert":
- to_txt = " ".join(after_sents[j1:j2]).strip()
- changes.append({"type": "insert", "from": "", "to": to_txt})
- for j in range(j1, j2):
- parts.append(
- f'{_escape_html(after_sents[j])}'
- )
- elif tag == "delete":
- from_txt = " ".join(before_sents[i1:i2]).strip()
- changes.append({"type": "delete", "from": from_txt, "to": ""})
- else:
- # Defensive: should not happen.
- for j in range(j1, j2):
- parts.append(_escape_html(after_sents[j]))
-
- diff_html = " ".join(parts).strip()
- return diff_html, changes
-
-
-def _diff_title_html(before_title: str, after_title: str) -> Tuple[str, List[Dict[str, str]]]:
- before_t = (before_title or "").strip()
- after_t = (after_title or "").strip()
- if before_t == after_t:
- return "", []
- if not after_t:
- # Title removed: show nothing in UI, but keep "from/to" for debug.
- return "", [{"type": "delete", "from": before_t, "to": ""}]
- return (
- f'{_escape_html(after_t)}',
- [{"type": "replace", "from": before_t, "to": after_t}],
- )
-
-
-def _max_sentences_for_level(cascade_level: int, operation: str) -> int:
- if operation == "insert":
- return 2
- if cascade_level <= 1:
- return 2
- if cascade_level == 2:
- return 3
- return 4
-
-
-def _phrase_strategy_variants(mode: str, goal_type: str, goal_label: str) -> List[str]:
- normalized = (mode or "auto").strip().lower()
- if normalized == "ensemble":
- if goal_type != "bert":
- return ["auto"]
- phrase_len = len(_tokenize(goal_label or ""))
- if phrase_len >= 3:
- return ["distributed_preferred", "auto", "exact_preferred"]
- return ["auto", "exact_preferred", "distributed_preferred"]
- if normalized in {"auto", "exact_preferred", "distributed_preferred"}:
- return [normalized]
- return ["auto"]
-
-
-def _build_phrase_strategy_plan(mode: str, goal_type: str, goal_label: str, count: int) -> List[str]:
- variants = _phrase_strategy_variants(mode, goal_type, goal_label)
- if count <= 0:
- return []
- return [variants[i % len(variants)] for i in range(count)]
-
-
-def _validate_title_candidate(edited_text: str) -> List[str]:
- """Guardrails for HTML plain text (not body sentences)."""
- reasons: List[str] = []
- text = (edited_text or "").strip()
- if not text:
- reasons.append("empty_title")
- return reasons
- if "\n" in text or "\r" in text:
- reasons.append("title_multiline")
- if len(text) > 90:
- reasons.append("title_too_long>90")
- if len(text) < 8:
- reasons.append("title_too_short")
- if re.search(r"<[^>]+>", text):
- reasons.append("title_contains_html_tags")
- return reasons
-
-
-def _validate_candidate_text(
- edited_text: str,
- cascade_level: int,
- operation: str,
- goal_label: str = "",
- focus_terms: Optional[List[str]] = None,
-) -> List[str]:
- reasons: List[str] = []
- text = (edited_text or "").strip()
- if not text:
- reasons.append("empty_candidate")
- return reasons
-
- sentence_count = len(_split_sentences(text))
- max_sent = _max_sentences_for_level(cascade_level, operation)
- if sentence_count > max_sent:
- reasons.append(f"too_many_sentences>{max_sent}")
-
- # Heuristic quality checks: duplicated words/entities and obvious malformed token joins.
- if re.search(r"\b([A-Za-z][A-Za-z0-9-]{1,})\s+\1\b", text, flags=re.IGNORECASE):
- reasons.append("duplicated_entity_or_word")
- # Catch broken lowercase+Camel join artifacts like "likemBit", but allow brand CamelCase like "RedDogCasino".
- if re.search(r"\b[a-z]{6,}[A-Z][a-z]+\b", text):
- reasons.append("suspicious_token_join")
-
- # Anti-stuffing checks for BERT phrase goals.
- focus_terms = focus_terms or []
- phrase = (goal_label or "").strip().lower()
- normalized = re.sub(r"\s+", " ", text.lower())
-
- if phrase:
- phrase_occurrences = normalized.count(phrase)
- phrase_token_count = len(_tokenize(phrase))
- # For long goal phrases, repeated exact matches are usually unnatural.
- if phrase_token_count >= 3 and phrase_occurrences > 1:
- reasons.append("exact_phrase_stuffing")
- elif phrase_occurrences > 2:
- reasons.append("exact_phrase_stuffing")
-
- for term in focus_terms:
- tok = (term or "").strip().lower()
- if not tok:
- continue
- term_occurrences = len(re.findall(rf"\b{re.escape(tok)}\b", normalized))
- if term_occurrences > 3:
- reasons.append("focus_term_overuse")
- break
-
- return reasons
-
-
-def _build_analysis_snapshot(
- target_text: str,
- competitors: List[str],
- keywords: List[str],
- language: str,
- target_title: str,
- competitor_titles: List[str],
-) -> Dict[str, Any]:
- wc_target = logic.count_words(target_text, language)
- wc_comp = [logic.count_words(t, language) for t in competitors]
- if wc_comp:
- avg_total = sum(c["total"] for c in wc_comp) / len(wc_comp)
- avg_sig = sum(c["significant"] for c in wc_comp) / len(wc_comp)
- else:
- avg_total = 0
- avg_sig = 0
-
- ngram_stats = logic.calculate_ngram_stats(target_text, competitors, language)
- key_phrases, _ = logic.parse_keywords(keywords, language)
- bm25 = logic.calculate_bm25_recommendations(target_text, competitors, keywords, language)
- bert = logic.perform_bert_analysis(target_text, competitors, key_phrases, language)
-
- title_data = {}
- if (target_title or "").strip():
- title_data = logic.analyze_title(target_title, competitor_titles, keywords, language)
-
- return {
- "word_counts": {
- "target": wc_target,
- "competitors": wc_comp,
- "avg": {"total": round(avg_total), "significant": round(avg_sig)},
- },
- "ngram_stats": ngram_stats,
- "bm25_recommendations": bm25,
- "bert_analysis": bert,
- "title_analysis": title_data,
- }
-
-
-def _build_semantic_snapshot(
- target_text: str,
- competitors: List[str],
- language: str,
-) -> Dict[str, Any]:
- def _build_doc(text: str, doc_id: int) -> Dict[str, Any]:
- sentences_data = nlp_processor.preprocess_text(text, language)
- graph, word_weights = semantic_graph.build_semantic_graph(sentences_data, lang=language)
- graph_data = semantic_graph.get_graph_data_for_frontend(graph)
- return {
- "id": doc_id,
- "text": text,
- "word_weights": word_weights,
- "stats": {
- "nodes": len(graph_data.get("nodes", [])),
- "links": len(graph_data.get("links", [])),
- },
- }
-
- target_doc = _build_doc(target_text, 0)
- comp_docs = []
- for idx, c in enumerate([x for x in competitors if (x or "").strip()]):
- comp_docs.append(_build_doc(c, idx + 1))
-
- num_comp = len(comp_docs)
- target_weights = target_doc["word_weights"]
- all_terms = set(target_weights.keys())
- for c in comp_docs:
- all_terms.update(c["word_weights"].keys())
-
- term_power_table = []
- for term in all_terms:
- target_weight = int(target_weights.get(term, 0))
- comp_weights = [int(c["word_weights"].get(term, 0)) for c in comp_docs]
- comp_avg = round(sum(comp_weights) / max(1, num_comp), 2)
- comp_occ = sum(1 for w in comp_weights if w > 0)
- term_power_table.append(
- {
- "term": term,
- "target_weight": target_weight,
- "competitor_avg_weight": comp_avg,
- "comp_occurrence": comp_occ,
- "comp_total": num_comp,
- }
- )
- return {"comparison": {"term_power_table": term_power_table, "num_competitors": num_comp}}
-
-
-def _is_semantic_gap(target_weight: float, competitor_avg_weight: float) -> bool:
- # Gap is significant only when competitor is meaningfully above target:
- # - relative margin above tolerance band (15%)
- # - and absolute margin to avoid micro-noise around normalized weights
- if competitor_avg_weight <= 0:
- return False
- abs_gap = competitor_avg_weight - target_weight
- rel_threshold = target_weight * (1.0 + SEMANTIC_GAP_TOLERANCE_PCT)
- return (competitor_avg_weight > rel_threshold) and (abs_gap >= SEMANTIC_GAP_MIN_ABS)
-
-
-def _ngram_tolerance_pct(competitor_avg: float) -> float:
- # User rule:
- # - avg >= 4 -> +/-20%
- # - avg < 4 -> +/-50%
- return 0.20 if competitor_avg >= 4.0 else 0.50
-
-
-def _is_ngram_outside_tolerance(target_count: float, competitor_avg: float) -> bool:
- if competitor_avg <= 0:
- return False
- tol = _ngram_tolerance_pct(competitor_avg)
- low = competitor_avg * (1.0 - tol)
- high = competitor_avg * (1.0 + tol)
- return target_count < low or target_count > high
-
-
-def _ngram_deviation_ratio(target_count: float, competitor_avg: float) -> float:
- if competitor_avg <= 0:
- return 0.0
- return abs(target_count - competitor_avg) / max(competitor_avg, 1e-6)
-
-
-def _keyword_unigram_set(keywords: List[str], language: str) -> set:
- out = set()
- for kw in keywords:
- toks = _filter_stopwords(_tokenize(kw), language)
- for t in toks:
- out.add(t)
- return out
-
-
-def _is_ngram_stage_candidate(
- ngram_label: str,
- comp_occurrence: int,
- competitor_count: int,
- keyword_unigrams: set,
-) -> bool:
- ngram = (ngram_label or "").strip().lower()
- if not ngram:
- return False
- tokens = _tokenize(ngram)
- n = len(tokens)
- if competitor_count > 1:
- if comp_occurrence < 2:
- return False
- if n >= 2:
- # For multi-competitor mode, bi/tri-grams with K>=2 are always candidates.
- return True
- # Unigrams are candidates only if they belong to key phrases.
- return n == 1 and tokens[0] in keyword_unigrams
- # Single-competitor mode: keep broader eligibility.
- return comp_occurrence >= 1
-
-
-def _chunk_ngram_count(text: str, ngram_label: str, language: str) -> int:
- toks = _filter_stopwords(_tokenize(text), language)
- phrase_toks = _filter_stopwords(_tokenize(ngram_label), language)
- if not toks or not phrase_toks:
- return 0
- n = len(phrase_toks)
- if n == 1:
- term = phrase_toks[0]
- return sum(1 for t in toks if t == term)
- count = 0
- for i in range(0, max(0, len(toks) - n + 1)):
- if toks[i : i + n] == phrase_toks:
- count += 1
- return count
-
-
-def _build_ngram_stage_rows(
- analysis: Dict[str, Any],
- keywords: List[str],
- language: str,
-) -> List[Tuple[str, float, float, float, int, float]]:
- """
- Eligible underrepresented n-gram targets for the ngram stage.
- Each row: (ngram_label, target_count, comp_avg, tolerance_pct, comp_occurrence, dev_ratio).
-
- Priority (user policy): maximize competitor coverage Freq(K), then Avg(K), then
- how far below the band the target is (larger deviation first).
- Only terms with My=0 or outside the tolerance band below average are included.
- """
- ngram_rows: List[Tuple[str, float, float, float, int, float]] = []
- ngram_stats = analysis.get("ngram_stats", {}) or {}
- competitor_count = len((analysis.get("word_counts", {}) or {}).get("competitors", []) or [])
- keyword_unigrams = _keyword_unigram_set(keywords, language)
- for _bucket_name, bucket in ngram_stats.items():
- if not isinstance(bucket, list):
- continue
- for item in bucket:
- ngram_label = str(item.get("ngram", "")).strip()
- if not ngram_label:
- continue
- target = float(item.get("target_count", 0))
- comp_avg = float(item.get("competitor_avg", 0))
- comp_occ = int(item.get("comp_occurrence", 0))
- if not _is_ngram_stage_candidate(ngram_label, comp_occ, competitor_count, keyword_unigrams):
- continue
- if not _is_ngram_outside_tolerance(target, comp_avg):
- continue
- if target >= comp_avg:
- continue
- tol = _ngram_tolerance_pct(comp_avg)
- dev_ratio = _ngram_deviation_ratio(target, comp_avg)
- ngram_rows.append((ngram_label, target, comp_avg, tol, comp_occ, dev_ratio))
- ngram_rows.sort(key=lambda x: (x[4], x[2], x[5]), reverse=True)
- return ngram_rows
-
-
-def _score_ngram_candidate_window(window_sentences: List[str], goal_label: str, language: str) -> float:
- """Heuristic: good place to add phrase — low local duplication, topical proximity, not boilerplate."""
- chunk = " ".join(s for s in window_sentences if s).strip()
- if not chunk:
- return -1e6
- phrase_count = float(_chunk_ngram_count(chunk, goal_label, language))
- noise_n = sum(1 for s in window_sentences if _is_noise_like_sentence(s))
- noise_frac = noise_n / max(1, len(window_sentences))
- phrase_tokens = [t.lower() for t in _filter_stopwords(_tokenize(goal_label), language) if t]
- chunk_l = chunk.lower()
- unigram_hits = sum(1 for t in phrase_tokens if t and len(t) > 1 and t in chunk_l)
- rel_proxy = unigram_hits / max(1, len(phrase_tokens)) if phrase_tokens else 0.0
- return (
- -3.0 * phrase_count
- + 2.2 * rel_proxy
- - 4.0 * noise_frac
- + min(len(chunk) / 1200.0, 0.35)
- )
-
-
-def _rank_ngram_overlap_sentence_indices(
- sentences: List[str],
- goal_label: str,
- language: str,
-) -> List[int]:
- """
- Slide overlapping multi-sentence windows over the document; each sentence gets the
- best score among windows that contain it. Order sentences by that score (desc).
- """
- n = len(sentences)
- if n <= 0:
- return [0]
- if n == 1:
- return [0]
- # 2–4 sentences per window, stride 1 for strong overlap.
- w = min(4, max(2, n))
- best: List[float] = [-1e9] * n
- for start in range(0, n - w + 1):
- win = sentences[start : start + w]
- sc = _score_ngram_candidate_window(win, goal_label, language)
- for j in range(start, start + w):
- if sc > best[j]:
- best[j] = sc
- center = (n - 1) / 2.0
- scored_idx = [(i, best[i], -abs(i - center)) for i in range(n)]
- scored_idx.sort(key=lambda t: (t[1], t[2]), reverse=True)
- return [t[0] for t in scored_idx]
-
-
-def _compute_metrics(
- analysis: Dict[str, Any],
- semantic: Dict[str, Any],
- keywords: List[str],
- language: str,
- bert_stage_target: float = BERT_TARGET_THRESHOLD,
-) -> Dict[str, Any]:
- competitor_count = len(analysis.get("word_counts", {}).get("competitors", []))
- min_signal = 1 if competitor_count <= 1 else 2
-
- bert_details = analysis.get("bert_analysis", {}).get("detailed", []) or []
- bert_low = [d for d in bert_details if float(d.get("my_max_score", 0)) < float(bert_stage_target)]
- bert_phrase_scores = {
- str(d.get("phrase", "")).strip().lower(): float(d.get("my_max_score", 0) or 0.0)
- for d in bert_details
- if str(d.get("phrase", "")).strip()
- }
-
- bm25_remove = [x for x in (analysis.get("bm25_recommendations") or []) if x.get("action") == "remove"]
- bm25_remove_count = len(bm25_remove)
-
- ngram_signal_count = 0
- ngram_gap_sum = 0.0
- keyword_unigrams = _keyword_unigram_set(keywords, language)
- ngrams = analysis.get("ngram_stats", {}) or {}
- for bucket in ngrams.values():
- if not isinstance(bucket, list):
- continue
- for item in bucket:
- comp_occ = int(item.get("comp_occurrence", 0))
- ngram_label = str(item.get("ngram", ""))
- if not _is_ngram_stage_candidate(ngram_label, comp_occ, competitor_count, keyword_unigrams):
- continue
- target = float(item.get("target_count", 0))
- comp_avg = float(item.get("competitor_avg", 0))
- if _is_ngram_outside_tolerance(target, comp_avg):
- ngram_signal_count += 1
- ngram_gap_sum += _ngram_deviation_ratio(target, comp_avg)
-
- title_score = None
- title_bert = analysis.get("title_analysis", {}).get("bert", {})
- if title_bert and title_bert.get("target_score") is not None:
- title_score = float(title_bert.get("target_score", 0))
-
- keyword_terms = set()
- for kw in keywords:
- tokens = _filter_stopwords(_tokenize(kw), language)
- for t in tokens:
- keyword_terms.add(t)
- for n in (2, 3):
- for i in range(0, max(0, len(tokens) - n + 1)):
- keyword_terms.add(" ".join(tokens[i : i + n]))
-
- table = semantic.get("comparison", {}).get("term_power_table", []) or []
- by_term = {str(r.get("term", "")).lower(): r for r in table}
- semantic_gap_count = 0
- semantic_gap_sum = 0.0
- semantic_gap_terms: List[Dict[str, Any]] = []
- for term in keyword_terms:
- row = by_term.get(term)
- if not row:
- continue
- target_w = float(row.get("target_weight", 0))
- comp_w = float(row.get("competitor_avg_weight", 0))
- gap = comp_w - target_w
- if int(row.get("comp_occurrence", 0)) >= min_signal and _is_semantic_gap(target_w, comp_w):
- semantic_gap_count += 1
- semantic_gap_sum += gap
- base = max(1.0, target_w)
- semantic_gap_terms.append(
- {
- "term": term,
- "target_weight": round(target_w, 2),
- "competitor_avg_weight": round(comp_w, 2),
- "gap": round(gap, 2),
- "gap_pct_of_target": round(gap / base, 4),
- "comp_occurrence": int(row.get("comp_occurrence", 0)),
- "comp_total": int(row.get("comp_total", 0)),
- }
- )
-
- # Composite score (0..100)
- w_bert, w_bm25, w_ng, w_title, w_sem = 30, 20, 15, 10, 25
- bert_comp = 1.0 - (len(bert_low) / max(1, len(bert_details)))
- bm25_comp = 1.0 if bm25_remove_count <= 3 else max(0.0, 1.0 - ((bm25_remove_count - 3) / 10.0))
- ng_comp = max(0.0, 1.0 - (ngram_signal_count / 15.0))
- title_comp = 1.0 if title_score is None else min(1.0, max(0.0, title_score / 0.65))
- sem_comp = max(0.0, 1.0 - (semantic_gap_count / 20.0))
-
- weighted = (
- w_bert * bert_comp
- + w_bm25 * bm25_comp
- + w_ng * ng_comp
- + w_title * title_comp
- + w_sem * sem_comp
- )
- total_w = w_bert + w_bm25 + w_ng + w_title + w_sem
- score = round((weighted / total_w) * 100.0, 2)
-
- resolved_title = ""
- _td = analysis.get("title_analysis") or {}
- if isinstance(_td, dict) and (_td.get("target_title") or "").strip():
- resolved_title = str(_td.get("target_title")).strip()
-
- return {
- "score": score,
- "competitor_count": competitor_count,
- "min_competitor_signal": min_signal,
- "bert_low_count": len(bert_low),
- "bert_total_keywords": len(bert_details),
- "bert_phrase_scores": bert_phrase_scores,
- "bm25_remove_count": bm25_remove_count,
- "ngram_signal_count": ngram_signal_count,
- "ngram_gap_sum": round(ngram_gap_sum, 4),
- "title_bert_score": title_score,
- "resolved_title": resolved_title,
- "semantic_gap_count": semantic_gap_count,
- "semantic_gap_sum": round(semantic_gap_sum, 4),
- "semantic_gap_terms": sorted(
- semantic_gap_terms,
- key=lambda x: (x.get("gap", 0), x.get("gap_pct_of_target", 0)),
- reverse=True,
- )[:20],
- }
-
-
-def _choose_optimization_goal(
- analysis: Dict[str, Any],
- semantic: Dict[str, Any],
- keywords: List[str],
- language: str,
- stage: str = "bert",
- bert_stage_target: float = BERT_TARGET_THRESHOLD,
- stage_cursor: int = 0,
-) -> Dict[str, Any]:
- goals = _collect_optimization_goals(
- analysis=analysis,
- semantic=semantic,
- keywords=keywords,
- language=language,
- stage=stage,
- bert_stage_target=bert_stage_target,
- )
- if not goals:
- return {"type": "none", "label": "no-op", "focus_terms": [], "avoid_terms": []}
- pick = max(0, int(stage_cursor))
- if pick >= len(goals):
- return {"type": "none", "label": "no-op", "focus_terms": [], "avoid_terms": []}
- return goals[pick]
-
-
-def _collect_optimization_goals(
- analysis: Dict[str, Any],
- semantic: Dict[str, Any],
- keywords: List[str],
- language: str,
- stage: str = "bert",
- bert_stage_target: float = BERT_TARGET_THRESHOLD,
- stage_goal_overrides: Optional[Dict[str, Any]] = None,
-) -> List[Dict[str, Any]]:
- goals: List[Dict[str, Any]] = []
- bert_details = analysis.get("bert_analysis", {}).get("detailed", []) or []
- low_bert = [x for x in bert_details if float(x.get("my_max_score", 0)) < float(bert_stage_target)]
- if low_bert:
- for row in sorted(low_bert, key=lambda x: float(x.get("my_max_score", 0))):
- phrase = str(row.get("phrase", "")).strip()
- if not phrase:
- continue
- focus_terms = _filter_stopwords(_tokenize(phrase), language)[:4]
- goals.append(
- {
- "type": "bert",
- "label": phrase,
- "focus_terms": focus_terms,
- "avoid_terms": [],
- "bert_phrase_score": float(row.get("my_max_score", 0) or 0.0),
- "bert_target": float(bert_stage_target),
- }
- )
-
- bm25_remove = [x for x in (analysis.get("bm25_recommendations") or []) if x.get("action") == "remove"]
- if len(bm25_remove) >= 4:
- for row in sorted(bm25_remove, key=lambda r: int(r.get("count", 0)), reverse=True)[:8]:
- word = str(row.get("word", "")).strip()
- if not word:
- continue
- goals.append(
- {
- "type": "bm25",
- "label": f"reduce spam: {word}",
- "focus_terms": [],
- "avoid_terms": [word],
- "bm25_count": int(row.get("count", 0) or 0),
- "bm25_word": word,
- }
- )
-
- # Semantic keyword gaps
- lang_stop = STOP_WORDS.get(language, STOP_WORDS["en"])
- keyword_terms = set()
- for kw in keywords:
- toks = [t for t in _tokenize(kw) if t not in lang_stop]
- keyword_terms.update(toks)
- for n in (2, 3):
- for i in range(0, max(0, len(toks) - n + 1)):
- keyword_terms.add(" ".join(toks[i : i + n]))
- table = semantic.get("comparison", {}).get("term_power_table", []) or []
- candidate_rows: List[Tuple[str, float]] = []
- for row in table:
- term = str(row.get("term", "")).lower()
- if term not in keyword_terms:
- continue
- target_w = float(row.get("target_weight", 0))
- comp_w = float(row.get("competitor_avg_weight", 0))
- gap = comp_w - target_w
- if _is_semantic_gap(target_w, comp_w):
- candidate_rows.append((term, gap))
- if candidate_rows:
- for term, gap in sorted(candidate_rows, key=lambda x: x[1], reverse=True)[:12]:
- goals.append(
- {
- "type": "semantic",
- "label": term,
- "focus_terms": [term],
- "avoid_terms": [],
- "semantic_gap": float(gap),
- }
- )
-
- # N-gram balancing (toward competitor average with tolerance policy).
- ngram_rows = _build_ngram_stage_rows(analysis, keywords, language)
- if ngram_rows:
- for rank, (label, target, comp_avg, tol, _, _) in enumerate(ngram_rows):
- goals.append({
- "type": "ngram",
- "label": label,
- "focus_terms": [label],
- "avoid_terms": [],
- "ngram_target_count": target,
- "ngram_comp_avg": comp_avg,
- "ngram_tolerance_pct": tol,
- "ngram_lower_bound": round(comp_avg * (1.0 - tol), 3),
- "ngram_upper_bound": round(comp_avg * (1.0 + tol), 3),
- "ngram_direction": "increase" if target < comp_avg else "decrease",
- "ngram_rank_index": rank,
- "ngram_candidates_total": len(ngram_rows),
- })
-
- title_bert = analysis.get("title_analysis", {}).get("bert", {}) or {}
- title_target_score = title_bert.get("target_score")
- if (
- keywords
- and title_target_score is not None
- and float(title_target_score) < TITLE_TARGET_THRESHOLD
- ):
- goals.append(
- {
- "type": "title",
- "label": "title alignment",
- "focus_terms": _filter_stopwords(_tokenize(" ".join(keywords[:8])), language)[:8],
- "avoid_terms": [],
- "title_bert_score": float(title_target_score) if title_target_score is not None else None,
- "title_target": float(TITLE_TARGET_THRESHOLD),
- }
- )
-
- stage_goals = [g for g in goals if g.get("type") == stage]
- return _apply_stage_goal_override(stage_goals, stage, language, stage_goal_overrides)
-
-
-def _per_goal_budget(
- goal: Dict[str, Any],
- max_iterations: int,
- candidates_per_iteration: int,
- bert_stage_target: float,
-) -> Tuple[int, int]:
- """
- Scale per-goal iteration and candidate budgets by how far the metric is from its target.
- Returns (effective_max_iterations_for_this_goal, effective_candidates_per_iteration).
- """
- t = str(goal.get("type", "") or "")
- raw = 0.0
-
- if t == "bert":
- sc = float(goal.get("bert_phrase_score", 0.0) or 0.0)
- tgt = float(goal.get("bert_target", bert_stage_target) or bert_stage_target)
- raw = max(0.0, (tgt - sc) / max(tgt, 1e-6))
- elif t == "ngram":
- ca = float(goal.get("ngram_comp_avg", 0.0) or 0.0)
- tc = float(goal.get("ngram_target_count", 0.0) or 0.0)
- if str(goal.get("ngram_direction", "increase")) == "increase":
- need = max(0.0, ca - tc)
- raw = min(1.0, need / max(ca, 1e-6))
- else:
- need = max(0.0, tc - ca)
- raw = min(1.0, need / max(tc, 1e-6))
- elif t == "semantic":
- gap = float(goal.get("semantic_gap", 0.0) or 0.0)
- raw = min(1.0, gap / max(SEMANTIC_GAP_MIN_ABS * 4.0, 1e-6))
- elif t == "bm25":
- c = int(goal.get("bm25_count", 0) or 0)
- raw = min(1.0, max(0, c - 1) / 8.0)
- elif t == "title":
- ts = goal.get("title_bert_score")
- if ts is None:
- raw = 0.5
- else:
- tgt = float(goal.get("title_target", TITLE_TARGET_THRESHOLD) or TITLE_TARGET_THRESHOLD)
- raw = max(0.0, (tgt - float(ts)) / max(tgt, 1e-6))
- else:
- raw = 0.0
-
- iter_mult = 1.0 + 2.0 * min(1.0, raw)
- cand_mult = 1.0 + 1.0 * min(1.0, raw)
- eff_iter = max(1, min(int(round(max_iterations * iter_mult)), max_iterations * 3))
- eff_cand = max(1, min(int(round(candidates_per_iteration * cand_mult)), 5))
- return eff_iter, eff_cand
-
-
-def _estimate_total_loop_budget(
- analysis: Dict[str, Any],
- semantic: Dict[str, Any],
- keywords: List[str],
- language: str,
- max_iterations: int,
- candidates_per_iteration: int,
- bert_stage_target: float,
- active_stage_order: Optional[List[str]] = None,
- stage_goal_overrides: Optional[Dict[str, Any]] = None,
-) -> int:
- total = 0
- stages = active_stage_order or list(STAGE_ORDER)
- for st in stages:
- for g in _collect_optimization_goals(
- analysis,
- semantic,
- keywords,
- language,
- stage=st,
- bert_stage_target=bert_stage_target,
- stage_goal_overrides=stage_goal_overrides,
- ):
- ei, _ = _per_goal_budget(g, max_iterations, candidates_per_iteration, bert_stage_target)
- total += ei
- return min(480, max(1, total))
-
-
-def _choose_sentence_idx(sentences: List[str], focus_terms: List[str], avoid_terms: List[str], language: str) -> int:
- if not sentences:
- return 0
- stop = STOP_WORDS.get(language, STOP_WORDS["en"])
- focus = [x for x in focus_terms if x and x not in stop]
-
- if avoid_terms:
- best_idx, best_score = 0, -1.0
- for i, s in enumerate(sentences):
- lower = s.lower()
- score = sum(lower.count(t.lower()) for t in avoid_terms if t)
- if score > best_score:
- best_idx, best_score = i, score
- return best_idx
-
- if focus:
- best_idx, best_score = 0, -1.0
- for i, s in enumerate(sentences):
- lower = s.lower()
- score = sum(lower.count(t.lower()) for t in focus)
- if score > best_score:
- best_idx, best_score = i, score
- return best_idx
-
- return min(2, len(sentences) - 1)
-
-
-def _rank_sentence_indices(
- sentences: List[str],
- focus_terms: List[str],
- avoid_terms: List[str],
- language: str,
- goal_type: str = "",
- goal_label: str = "",
-) -> List[int]:
- if not sentences:
- return [0]
- stop = STOP_WORDS.get(language, STOP_WORDS["en"])
- focus = [x for x in focus_terms if x and x not in stop]
- goal_phrase = (goal_label or "").strip().lower()
- avoid = [x for x in avoid_terms if x]
- center = (len(sentences) - 1) / 2.0
-
- # N-gram stage: overlapping sentence windows — pick spans where insertion is natural
- # while document-level phrase count remains the primary optimization signal.
- if goal_type == "ngram" and (goal_label or "").strip():
- return _rank_ngram_overlap_sentence_indices(sentences, str(goal_label).strip(), language)
-
- # For BERT optimization prefer natural prose chunks over list/menu/noisy blocks.
- candidate_indices = list(range(len(sentences)))
- if goal_type == "bert":
- non_noise = [i for i, s in enumerate(sentences) if not _is_noise_like_sentence(s)]
- if non_noise:
- candidate_indices = non_noise
-
- scored: List[Tuple[int, float, int]] = []
- for idx in candidate_indices:
- s = sentences[idx]
- lower = s.lower()
- # For bm25 "remove" goals we must select spans that actually contain the exact n-gram term.
- # Otherwise the LLM can "optimize around" the goal without decreasing the real BM25 remove metric.
- if goal_type == "bm25" and avoid and not focus_terms:
- bm25_term = avoid[0]
- bm25_count = _count_term_ngrams_strict(s, bm25_term)
- focus_score = 0
- avoid_score = float(bm25_count)
- chunk_rel = float(bm25_count)
- else:
- focus_score = sum(lower.count(t.lower()) for t in focus)
- avoid_score = sum(lower.count(t.lower()) for t in avoid)
- chunk_rel = _chunk_goal_relevance(s, goal_type, goal_label, focus_terms, language)
- noise_penalty = 1.0 if _is_noise_like_sentence(s) else 0.0
-
- # For BERT goals, do not over-focus on existing occurrences only:
- # prioritize semantically relevant chunks where phrase/terms may still be underrepresented.
- if goal_type == "bert":
- tokenized = _filter_stopwords(_tokenize(lower), language)
- token_set = set(tokenized)
- core_terms = [t.lower() for t in focus if t]
- core_hits = sum(1 for t in core_terms if t in token_set)
- coverage = (core_hits / max(1, len(core_terms))) if core_terms else 0.0
- phrase_present = 1.0 if (goal_phrase and goal_phrase in lower) else 0.0
- # Boost candidates where semantic context is relevant but explicit core terms are not saturated yet.
- missing_term_boost = (1.0 - coverage) * 1.4 if chunk_rel >= 0.18 else 0.0
- phrase_absent_boost = 0.35 if (goal_phrase and not phrase_present and chunk_rel >= 0.2) else 0.0
- score = (
- (chunk_rel * 5.0)
- + (focus_score * 1.2)
- + (missing_term_boost + phrase_absent_boost)
- + (avoid_score * 1.5)
- - (noise_penalty * 3.0)
- - (abs(idx - center) * 0.05)
- )
- else:
- # Prefer semantically relevant and lexical matches; push noisy headers/CTA lower.
- score = (chunk_rel * 4.0) + (focus_score * 3.0) + (avoid_score * 2.0) - (noise_penalty * 3.0) - (abs(idx - center) * 0.05)
- scored.append((idx, score, len(s)))
-
- scored.sort(key=lambda x: (x[1], -x[2]), reverse=True)
- ordered = [idx for idx, _, _ in scored]
-
- if not ordered:
- ordered = list(range(len(sentences)))
- return ordered
-
-
-def _span_variants_for_level(cascade_level: int) -> List[Tuple[str, int, int]]:
- # (operation, left_radius, right_radius)
- if cascade_level <= 1:
- return [("rewrite", 0, 0), ("rewrite", 0, 1), ("rewrite", 1, 0)]
- if cascade_level == 2:
- return [("rewrite", 1, 1), ("rewrite", 0, 2), ("rewrite", 2, 0), ("rewrite", 1, 2), ("rewrite", 2, 1)]
- if cascade_level == 3:
- return [("insert", 0, 0), ("insert", 0, 0), ("insert", 0, 0), ("rewrite", 1, 1)]
- return [("rewrite", 2, 2), ("rewrite", 1, 3), ("rewrite", 3, 1), ("rewrite", 2, 3), ("rewrite", 3, 2)]
-
-
-def _choose_edit_span(
- sentences: List[str],
- goal: Dict[str, Any],
- language: str,
- cascade_level: int,
- attempt_cursor: int,
-) -> Tuple[str, int, int, int, int]:
- ranked = _rank_sentence_indices(
- sentences,
- goal.get("focus_terms", []) or [],
- goal.get("avoid_terms", []) or [],
- language,
- str(goal.get("type", "") or ""),
- str(goal.get("label", "") or ""),
- )
- variants = _span_variants_for_level(cascade_level)
- total = max(1, len(ranked) * len(variants))
- pick = attempt_cursor % total
- sent_pick = pick % len(ranked)
- variant_pick = (pick // len(ranked)) % len(variants)
-
- sent_idx = ranked[sent_pick]
- operation, left_radius, right_radius = variants[variant_pick]
-
- if operation == "insert":
- span_start = sent_idx
- span_end = sent_idx
- else:
- span_start = max(0, sent_idx - left_radius)
- span_end = min(len(sentences) - 1, sent_idx + right_radius)
-
- return operation, span_start, span_end, sent_idx, variant_pick
-
-
-def _portfolio_span_trials(goal_type: str, cascade_level: int, eff_candidates: int) -> int:
- """Try more distinct spans for hard local goals without exploding candidate count."""
- t = (goal_type or "").strip().lower()
- base = 2 if cascade_level <= 2 else 3
- if t in {"bert", "ngram"}:
- base += 1
- if cascade_level >= 3 and t in {"bert", "ngram", "semantic"}:
- base += 1
- return max(1, min(5, max(base, min(3, int(eff_candidates or 1)))))
-
-
-def _local_prefilter_reasons(
- *,
- goal_type: str,
- focus_terms: List[str],
- edited_text: str,
- chunk_delta: float,
- before_rel: float,
- after_rel: float,
-) -> List[str]:
- """
- Cheap local gate before the expensive full document metric rebuild.
- It rejects only candidates that cannot plausibly help the current goal.
- """
- t = (goal_type or "").strip().lower()
- reasons: List[str] = []
- if t == "bm25" and chunk_delta <= 0:
- reasons.append("local_bm25_term_not_reduced")
- elif t == "ngram" and chunk_delta <= 0:
- reasons.append("local_ngram_not_closer_to_target")
- elif t == "bert" and after_rel < (before_rel - 0.003):
- reasons.append("local_bert_relevance_dropped")
- elif t == "semantic":
- edited_l = (edited_text or "").lower()
- has_focus = any(str(t).strip().lower() in edited_l for t in (focus_terms or []) if str(t).strip())
- if chunk_delta <= 0 and not has_focus:
- reasons.append("local_semantic_focus_missing")
- return reasons
-
-
-def _is_noise_like_sentence(text: str) -> bool:
- s = (text or "").strip()
- if not s:
- return True
- lower = s.lower()
- tokens = _tokenize(lower)
- if len(tokens) <= 2:
- return True
- if len(tokens) <= 8 and re.search(r"\b(play|explore|best|top|contact|login|signup|casino)\b", lower):
- return True
- if len(s) <= 90 and re.fullmatch(r"[A-Z0-9\s\-\|\:\.]+", s):
- return True
- if re.search(r"\b(best|top)\b.{0,20}\b(alternative|alternatives|casino|casinos)\b", lower) and len(tokens) <= 12:
- return True
- return False
-
-
-def _chunk_goal_relevance(
- text: str,
- goal_type: str,
- goal_label: str,
- focus_terms: List[str],
- language: str,
- goal_meta: Optional[Dict[str, Any]] = None,
-) -> float:
- chunk = (text or "").strip()
- if not chunk:
- return 0.0
- if goal_type == "bert" and (goal_label or "").strip():
- try:
- model = logic.get_bert_model()
- embeddings = model.encode([goal_label.strip(), chunk], convert_to_tensor=True)
- return float(logic.util.cos_sim(embeddings[0:1], embeddings[1:2])[0][0].item())
- except Exception:
- pass
- if goal_type == "ngram" and (goal_label or "").strip():
- return float(_chunk_ngram_count(chunk, goal_label, language))
-
- # Lexical fallback for non-BERT goals or if embedding scoring is unavailable.
- toks = _filter_stopwords(_tokenize(chunk), language)
- if not toks:
- return 0.0
- focus = [f.lower() for f in (focus_terms or []) if f]
- if not focus:
- return 0.0
- overlap = 0.0
- token_str = " ".join(toks)
- for f in focus:
- if " " in f:
- overlap += 1.0 if f in token_str else 0.0
- else:
- overlap += 1.0 if f in toks else 0.0
- return overlap / max(1.0, float(len(focus)))
-
-
-def _chunk_goal_delta(
- before_text: str,
- after_text: str,
- goal_type: str,
- goal_label: str,
- focus_terms: List[str],
- language: str,
- goal_meta: Optional[Dict[str, Any]] = None,
-) -> float:
- # BM25 "remove spam" must be judged by strict decrease of the exact n-gram term count.
- # Иначе LLM может улучшать семантику рядом, но не уменьшать реальный BM25 remove metric.
- if goal_type == "bm25":
- term = None
- avoid = []
- focus = []
- if isinstance(goal_meta, dict):
- term = goal_meta.get("bm25_word")
- avoid = goal_meta.get("avoid_terms") or []
- focus = goal_meta.get("focus_terms") or []
- # Fallback: try to infer from avoid_terms
- if not term and avoid:
- term = avoid[0]
- if not term:
- return 0.0
-
- before_count = _count_term_ngrams_strict(before_text or "", str(term))
- after_count = _count_term_ngrams_strict(after_text or "", str(term))
-
- # For now bm25 goals are "remove" (focus_terms is empty, avoid_terms contains term).
- # Keep the sign direction explicit so add-mode can be introduced later safely.
- if avoid and not focus:
- return round(float(before_count - after_count), 4)
- # If ever we add bm25 "add" goals with focus terms, then increase is good.
- return round(float(after_count - before_count), 4)
-
- before_rel = _chunk_goal_relevance(before_text, goal_type, goal_label, focus_terms, language, goal_meta)
- after_rel = _chunk_goal_relevance(after_text, goal_type, goal_label, focus_terms, language, goal_meta)
- if goal_type == "ngram":
- target_avg = float((goal_meta or {}).get("ngram_comp_avg", 0.0))
- if target_avg > 0:
- # Positive delta means closer to competitor average regardless of direction.
- before_dist = abs(before_rel - target_avg)
- after_dist = abs(after_rel - target_avg)
- return round(before_dist - after_dist, 4)
- return round(after_rel - before_rel, 4)
-
-
-def _min_chunk_delta(goal_type: str) -> float:
- if goal_type == "bert":
- return 0.01
- if goal_type == "title":
- return 0.005
- if goal_type == "ngram":
- # Require at least one occurrence-equivalent movement toward target zone.
- return 0.5
- return 0.05
-
-
-def _chunk_relevance_pair(
- before_text: str,
- after_text: str,
- goal_type: str,
- goal_label: str,
- focus_terms: List[str],
- language: str,
- goal_meta: Optional[Dict[str, Any]] = None,
-) -> Tuple[float, float]:
- before_rel = _chunk_goal_relevance(before_text, goal_type, goal_label, focus_terms, language, goal_meta)
- after_rel = _chunk_goal_relevance(after_text, goal_type, goal_label, focus_terms, language, goal_meta)
- return round(before_rel, 4), round(after_rel, 4)
-
-
-def _term_diff(before_text: str, after_text: str, language: str) -> Dict[str, List[str]]:
- before_tokens = _filter_stopwords(_tokenize(before_text), language)
- after_tokens = _filter_stopwords(_tokenize(after_text), language)
- before_set = set(before_tokens)
- after_set = set(after_tokens)
- removed = sorted(list(before_set - after_set))[:12]
- added = sorted(list(after_set - before_set))[:12]
- return {"added_terms": added, "removed_terms": removed}
-
-
-def _edits_conflict(a: Dict[str, Any], b: Dict[str, Any]) -> bool:
- if a.get("operation") == "title_rewrite" or b.get("operation") == "title_rewrite":
- return True
- if a.get("operation") == "insert" or b.get("operation") == "insert":
- return int(a.get("span_start", 0)) == int(b.get("span_start", 0))
- a0, a1 = int(a.get("span_start", 0)), int(a.get("span_end", 0))
- b0, b1 = int(b.get("span_start", 0)), int(b.get("span_end", 0))
- return not (a1 < b0 or b1 < a0)
-
-
-def _apply_edits_to_sentences(sentences: List[str], edits: List[Dict[str, Any]]) -> List[str]:
- updated = list(sentences)
- ordered = sorted(
- edits,
- key=lambda e: (int(e.get("span_start", 0)), int(e.get("span_end", 0))),
- reverse=True,
- )
- for edit in ordered:
- op = str(edit.get("operation", "rewrite"))
- start = int(edit.get("span_start", 0))
- end = int(edit.get("span_end", start))
- text = str(edit.get("edited_text", "")).strip()
- if not text:
- continue
- if op == "insert":
- updated = _insert_after(updated, end, text)
- else:
- updated = _replace_span(updated, start, end, text)
- return updated
-
-
-def _metrics_delta(prev_metrics: Dict[str, Any], next_metrics: Dict[str, Any]) -> Dict[str, Any]:
- keys = [
- "score",
- "bert_low_count",
- "bm25_remove_count",
- "ngram_signal_count",
- "ngram_gap_sum",
- "semantic_gap_count",
- "semantic_gap_sum",
- ]
- out: Dict[str, Any] = {}
- for k in keys:
- pv = prev_metrics.get(k)
- nv = next_metrics.get(k)
- if pv is None or nv is None:
- continue
- out[k] = round(float(nv) - float(pv), 4)
- pv_t = prev_metrics.get("title_bert_score")
- nv_t = next_metrics.get("title_bert_score")
- if pv_t is not None and nv_t is not None:
- out["title_bert_score"] = round(float(nv_t) - float(pv_t), 4)
- prev_terms = {str(x.get("term", "")).lower() for x in (prev_metrics.get("semantic_gap_terms") or []) if x.get("term")}
- next_terms = {str(x.get("term", "")).lower() for x in (next_metrics.get("semantic_gap_terms") or []) if x.get("term")}
- if prev_terms or next_terms:
- out["semantic_gap_terms_added"] = sorted(list(next_terms - prev_terms))[:12]
- out["semantic_gap_terms_removed"] = sorted(list(prev_terms - next_terms))[:12]
- return out
-
-
-def _non_conflicting_edit_combos(candidates: List[Dict[str, Any]], min_size: int = 2, max_size: int = 4) -> List[List[Dict[str, Any]]]:
- if not candidates:
- return []
- n = len(candidates)
- combos: List[List[Dict[str, Any]]] = []
- for r in range(max(2, min_size), min(max_size, n) + 1):
- for idxs in combinations(range(n), r):
- combo = [candidates[i] for i in idxs]
- conflict = False
- for i in range(len(combo)):
- for j in range(i + 1, len(combo)):
- e1 = combo[i].get("edit_payload")
- e2 = combo[j].get("edit_payload")
- if not e1 or not e2:
- conflict = True
- break
- if _edits_conflict(e1, e2):
- conflict = True
- break
- if conflict:
- break
- if not conflict:
- combos.append(combo)
- return combos
-
-
-def _extract_json_object(text: str) -> Optional[Dict[str, Any]]:
- raw = (text or "").strip()
- if not raw:
- return None
- try:
- return json.loads(raw)
- except Exception:
- pass
- m = re.search(r"\{[\s\S]*\}", raw)
- if not m:
- return None
- try:
- return json.loads(m.group(0))
- except Exception:
- return None
-
-
-def _llm_edit_chunk(
- *,
- api_key: str,
- base_url: str,
- model: str,
- language: str,
- full_text: str,
- chunk_text: str,
- operation: str,
- context_before: str,
- context_after: str,
- cascade_level: int,
- goal_type: str,
- goal_label: str,
- focus_terms: List[str],
- avoid_terms: List[str],
- temperature: float,
- phrase_strategy_mode: str = "auto",
-) -> Dict[str, Any]:
- # Title optimization must never touch body text; see optimize_text branch for goal.type == "title".
- if str(goal_type or "").strip().lower() == "title":
- raise ValueError(
- "Internal error: goal_type 'title' must be handled via _llm_edit_title(), not _llm_edit_chunk(). "
- "If you see this on a server, deploy the current optimizer (title-only LLM path)."
- )
- endpoint = base_url.rstrip("/") + "/chat/completions"
- op = operation if operation in {"rewrite", "insert"} else "rewrite"
- system_msg = (
- "You are a semantic-vector optimizer for SEO tasks. "
- "Your task is to improve chunk relevance to the focus terms/goal phrase with minimal local edits. "
- "Preserve narrative flow, factual tone, and language. "
- "Return strict JSON only: {\"edited_text\": \"...\", \"rationale\": \"...\"}. "
- "Do not rewrite the whole text. Never change topic or introduce unrelated entities."
- )
- op_instruction = (
- "Rewrite the provided chunk only."
- if op == "rewrite"
- else "Create a short bridge chunk (1-2 sentences) to insert after the chunk."
- )
- max_sent = _max_sentences_for_level(cascade_level, op)
- phrase_tokens = _filter_stopwords(_tokenize(goal_label or ""), language)
- phrase_len = len(phrase_tokens)
- strategy_mode = (phrase_strategy_mode or "auto").strip().lower()
- if strategy_mode not in {"auto", "exact_preferred", "distributed_preferred", "ensemble"}:
- strategy_mode = "auto"
- if strategy_mode == "exact_preferred":
- phrase_strategy = (
- "Prefer one natural exact phrase mention when grammatically correct; otherwise use distributed core-term coverage."
- )
- elif strategy_mode == "distributed_preferred":
- phrase_strategy = (
- "Prefer distributed semantic coverage: spread core terms/lemmas naturally and avoid exact phrase unless absolutely natural."
- )
- elif phrase_len >= 3:
- phrase_strategy = (
- "Prefer distributed semantic coverage for long phrases: naturally spread core terms/lemmas across the local paragraph. "
- "Use exact phrase only if it is grammatically natural."
- )
- elif phrase_len == 2:
- phrase_strategy = (
- "For two-term goals, use either one natural exact phrase or distributed use of both terms without repetition."
- )
- else:
- phrase_strategy = (
- "For single-term goals, improve relevance using natural lexical variants and nearby semantic anchors."
- )
- user_msg = (
- f"Language: {language}\n"
- f"Operation: {op}\n"
- f"Cascade level: L{cascade_level}\n"
- f"Goal: {goal_type} ({goal_label})\n"
- f"Goal token count (without stopwords): {phrase_len}\n"
- f"Instruction: {op_instruction}\n"
- f"Must preserve overall narrative and style.\n"
- "Text must be grammatically correct and natural for native readers.\n"
- "Keep edits tightly local to the provided chunk and immediate context only.\n"
- "Edit must be substantive (not just synonyms) and should increase relevance to the goal phrase.\n"
- "Do not change the sentence subject/entity focus unless absolutely required by grammar.\n"
- f"Phrase strategy: {phrase_strategy}\n"
- f"Focus terms to strengthen: {', '.join(focus_terms) if focus_terms else '-'}\n"
- f"Terms to de-emphasize/avoid overuse: {', '.join(avoid_terms) if avoid_terms else '-'}\n\n"
- f"Chunk to edit/expand:\n{chunk_text}\n\n"
- f"Prev context:\n{context_before}\n\n"
- f"Next context:\n{context_after}\n\n"
- "Constraints:\n"
- "1) Keep text concise and locally coherent.\n"
- "2) Keep local coherence with surrounding text.\n"
- f"3) Max {max_sent} sentence(s) in edited_text.\n"
- "4) Keep key named entities from the original chunk unchanged when possible.\n"
- "5) For BERT goals, prioritize semantic alignment over exact phrase repetition.\n"
- "6) If exact phrase sounds unnatural, do NOT force it; use grammatically correct distributed wording.\n"
- "7) Exact phrase may appear at most once, and only when it reads naturally.\n"
- "8) Avoid repeating the same focus term more than needed; no stuffing.\n"
- "9) For rewrite: preserve original meaning sentence-by-sentence while improving relevance.\n"
- "10) Provide rationale in one short sentence.\n"
- "11) Only output JSON object."
- )
- payload = {
- "model": model,
- "temperature": float(max(0.0, min(1.2, temperature))),
- "messages": [
- {"role": "system", "content": system_msg},
- {"role": "user", "content": user_msg},
- ],
- "response_format": {"type": "json_object"},
- }
- headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
- response = requests.post(endpoint, headers=headers, json=payload, timeout=60)
- response.raise_for_status()
- data = response.json()
- content = (
- data.get("choices", [{}])[0]
- .get("message", {})
- .get("content", "")
- )
- parsed = _extract_json_object(content)
- edited = ""
- rationale = ""
- if parsed:
- edited = str(parsed.get("edited_text") or parsed.get("revised_sentence") or parsed.get("rewrite") or "").strip()
- rationale = str(parsed.get("rationale") or parsed.get("why") or "").strip()
- if not edited:
- raise ValueError("LLM returned invalid JSON edit payload.")
- return {
- "edited_text": edited,
- "rationale": rationale,
- "prompt_debug": {
- "operation": op,
- "cascade_level": cascade_level,
- "goal_type": goal_type,
- "goal_label": goal_label,
- "focus_terms": focus_terms,
- "avoid_terms": avoid_terms,
- "phrase_strategy_mode": strategy_mode,
- "goal_token_count": phrase_len,
- "phrase_strategy": phrase_strategy,
- "max_sentences": max_sent,
- "chunk_text": chunk_text,
- "context_before": context_before,
- "context_after": context_after,
- "temperature": float(max(0.0, min(1.2, temperature))),
- },
- }
-
-
-def _llm_edit_title(
- *,
- api_key: str,
- base_url: str,
- model: str,
- language: str,
- current_title: str,
- body_excerpt: str,
- competitor_title_hint: str,
- focus_terms: List[str],
- avoid_terms: List[str],
- keywords: List[str],
- cascade_level: int,
- temperature: float,
- phrase_strategy_mode: str = "auto",
-) -> Dict[str, Any]:
- """
- Rewrite only the page <title> plain text. Backend metric: mean BERT cos-sim(title, each keyword).
- """
- endpoint = base_url.rstrip("/") + "/chat/completions"
- strategy_mode = (phrase_strategy_mode or "auto").strip().lower()
- if strategy_mode not in {"auto", "exact_preferred", "distributed_preferred", "ensemble"}:
- strategy_mode = "auto"
- phrase_tokens = _filter_stopwords(_tokenize(" ".join(keywords[:6])), language)
- phrase_len = len(phrase_tokens)
- if strategy_mode == "exact_preferred":
- phrase_strategy = (
- "Prefer one natural exact phrase from the keyword list when it fits; otherwise distribute core terms."
- )
- elif strategy_mode == "distributed_preferred":
- phrase_strategy = "Prefer natural distribution of core terms; avoid stiff exact-match phrasing."
- elif phrase_len >= 3:
- phrase_strategy = (
- "Spread core lemmas across the title; exact multi-word match only if it reads like a real title."
- )
- elif phrase_len == 2:
- phrase_strategy = "Use both core ideas in one short title line without repetition."
- else:
- phrase_strategy = "Keep the title specific and aligned with the keyword theme using natural wording."
-
- system_msg = (
- "You optimize HTML <title> tag text for SEO. "
- "Output JSON only: {\"edited_text\": \"...\", \"rationale\": \"...\"}. "
- "edited_text must be the new title line ONLY: plain text, no <title> tags, no quotes wrapping the whole title. "
- "Do not change the article body. Preserve the primary brand/site name from the current title when present."
- )
- kw_line = ", ".join(k.strip() for k in (keywords or [])[:24] if k.strip()) or "-"
- user_msg = (
- f"Language: {language}\n"
- f"Cascade level: L{cascade_level}\n"
- f"Task: Improve semantic alignment of the TITLE with these keywords (average embedding similarity should rise).\n"
- f"Keywords (priority): {kw_line}\n"
- f"Focus terms: {', '.join(focus_terms) if focus_terms else '-'}\n"
- f"De-emphasize: {', '.join(avoid_terms) if avoid_terms else '-'}\n"
- f"Phrase strategy: {phrase_strategy}\n\n"
- f"Current title:\n{current_title.strip()}\n\n"
- f"Page content excerpt (context only; do not paste into title):\n{(body_excerpt or '')[:900]}\n\n"
- f"{competitor_title_hint}\n\n"
- "Constraints:\n"
- "1) Length about 35–60 characters when possible; hard max 88 characters.\n"
- "2) One line; no line breaks; title case or sentence case per language norms.\n"
- "3) Natural human phrasing; no keyword stuffing; each important term at most once unless grammar needs it.\n"
- "4) Do not add unrelated entities or claims.\n"
- "5) Rationale: one short sentence.\n"
- )
- payload = {
- "model": model,
- "temperature": float(max(0.0, min(1.2, temperature))),
- "messages": [
- {"role": "system", "content": system_msg},
- {"role": "user", "content": user_msg},
- ],
- "response_format": {"type": "json_object"},
- }
- headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
- response = requests.post(endpoint, headers=headers, json=payload, timeout=60)
- response.raise_for_status()
- data = response.json()
- content = (
- data.get("choices", [{}])[0]
- .get("message", {})
- .get("content", "")
- )
- parsed = _extract_json_object(content)
- edited = ""
- rationale = ""
- if parsed:
- edited = str(parsed.get("edited_text") or parsed.get("title") or "").strip()
- rationale = str(parsed.get("rationale") or "").strip()
- if not edited:
- raise ValueError("LLM returned invalid JSON title payload.")
- return {
- "edited_text": edited,
- "rationale": rationale,
- "prompt_debug": {
- "operation": "title_rewrite",
- "cascade_level": cascade_level,
- "goal_type": "title",
- "goal_label": "title alignment",
- "focus_terms": focus_terms,
- "avoid_terms": avoid_terms,
- "phrase_strategy_mode": strategy_mode,
- "phrase_strategy": phrase_strategy,
- "chunk_text": current_title.strip(),
- "context_before": (body_excerpt or "")[:500],
- "context_after": competitor_title_hint[:500] if competitor_title_hint else "",
- "temperature": float(max(0.0, min(1.2, temperature))),
- },
- }
-
-
-def _replace_span(sentences: List[str], start_idx: int, end_idx: int, replacement_text: str) -> List[str]:
- replacement = _split_sentences(replacement_text)
- if not replacement:
- replacement = [replacement_text.strip()]
- return sentences[:start_idx] + replacement + sentences[end_idx + 1 :]
-
-
-def _insert_after(sentences: List[str], after_idx: int, inserted_text: str) -> List[str]:
- insertion = _split_sentences(inserted_text)
- if not insertion:
- insertion = [inserted_text.strip()]
- return sentences[: after_idx + 1] + insertion + sentences[after_idx + 1 :]
-
-
-def _goal_improved(
- goal_type: str,
- goal_label: str,
- prev_metrics: Dict[str, Any],
- next_metrics: Dict[str, Any],
-) -> bool:
- if goal_type == "bert":
- key = (goal_label or "").strip().lower()
- prev_phrase = float((prev_metrics.get("bert_phrase_scores") or {}).get(key, 0.0))
- next_phrase = float((next_metrics.get("bert_phrase_scores") or {}).get(key, 0.0))
- # Accept smaller but real phrase-level gains to accumulate progress toward threshold (e.g. 0.51 -> 0.70).
- return (next_phrase - prev_phrase) >= BERT_GOAL_DELTA_MIN or next_metrics["bert_low_count"] < prev_metrics["bert_low_count"]
- if goal_type == "bm25":
- return next_metrics["bm25_remove_count"] < prev_metrics["bm25_remove_count"]
- if goal_type == "semantic":
- return next_metrics["semantic_gap_count"] < prev_metrics["semantic_gap_count"]
- if goal_type == "ngram":
- return (
- next_metrics["ngram_signal_count"] < prev_metrics["ngram_signal_count"]
- or float(next_metrics.get("ngram_gap_sum", 0.0)) < float(prev_metrics.get("ngram_gap_sum", 0.0))
- )
- if goal_type == "title":
- pt = prev_metrics.get("title_bert_score")
- nt = next_metrics.get("title_bert_score")
- if pt is None or nt is None:
- return False
- return float(nt) > float(pt) or float(nt) >= TITLE_TARGET_THRESHOLD
- return next_metrics["score"] > prev_metrics["score"]
-
-
-def _bert_phrase_delta(goal_label: str, prev_metrics: Dict[str, Any], next_metrics: Dict[str, Any]) -> float:
- key = (goal_label or "").strip().lower()
- prev_phrase = float((prev_metrics.get("bert_phrase_scores") or {}).get(key, 0.0))
- next_phrase = float((next_metrics.get("bert_phrase_scores") or {}).get(key, 0.0))
- return round(next_phrase - prev_phrase, 4)
-
-
-def _safe_delta(prev_metrics: Dict[str, Any], next_metrics: Dict[str, Any], key: str) -> float:
- try:
- return float(next_metrics.get(key, 0.0)) - float(prev_metrics.get(key, 0.0))
- except Exception:
- return 0.0
-
-
-def _stage_primary_progress(stage: str, prev_metrics: Dict[str, Any], next_metrics: Dict[str, Any]) -> bool:
- if stage == "bert":
- prev_low = int(prev_metrics.get("bert_low_count", 0))
- next_low = int(next_metrics.get("bert_low_count", 0))
- if next_low < prev_low:
- return True
- prev_max = max([0.0] + [float(v) for v in (prev_metrics.get("bert_phrase_scores") or {}).values()])
- next_max = max([0.0] + [float(v) for v in (next_metrics.get("bert_phrase_scores") or {}).values()])
- return (next_max - prev_max) >= BERT_GOAL_DELTA_MIN
- if stage == "bm25":
- return int(next_metrics.get("bm25_remove_count", 0)) < int(prev_metrics.get("bm25_remove_count", 0))
- if stage == "semantic":
- return (
- int(next_metrics.get("semantic_gap_count", 0)) < int(prev_metrics.get("semantic_gap_count", 0))
- or float(next_metrics.get("semantic_gap_sum", 0.0)) < float(prev_metrics.get("semantic_gap_sum", 0.0))
- )
- if stage == "ngram":
- return (
- int(next_metrics.get("ngram_signal_count", 0)) < int(prev_metrics.get("ngram_signal_count", 0))
- or float(next_metrics.get("ngram_gap_sum", 0.0)) < float(prev_metrics.get("ngram_gap_sum", 0.0))
- )
- if stage == "title":
- pv = prev_metrics.get("title_bert_score")
- nv = next_metrics.get("title_bert_score")
- if pv is None or nv is None:
- return False
- return float(nv) > float(pv)
- return False
-
-
-def _is_stage_complete(stage: str, metrics: Dict[str, Any], bert_stage_target: float = BERT_TARGET_THRESHOLD) -> bool:
- if stage == "bert":
- # Complete only when all tracked BERT phrase scores meet the threshold.
- # This enforces per-phrase target behavior (no early exit on one strong phrase).
- scores = [float(v) for v in (metrics.get("bert_phrase_scores") or {}).values()]
- if not scores:
- return True
- return min(scores) >= float(bert_stage_target)
- if stage == "bm25":
- return int(metrics.get("bm25_remove_count", 0)) <= 3
- if stage == "semantic":
- return int(metrics.get("semantic_gap_count", 0)) <= 0
- if stage == "ngram":
- return int(metrics.get("ngram_signal_count", 0)) <= 0
- if stage == "title":
- score = metrics.get("title_bert_score")
- return (score is None) or (float(score) >= TITLE_TARGET_THRESHOLD)
- return True
-
-
-def _advance_ngram_term_cursor(cursor_state: Dict[str, Dict[str, int]], stage_key: str) -> None:
- state = cursor_state.get(stage_key) or {"term_index": 0, "attempt_count": 0}
- attempts = int(state.get("attempt_count", 0)) + 1
- term_index = int(state.get("term_index", 0))
- if attempts >= NGRAM_ATTEMPTS_PER_TERM:
- term_index += 1
- attempts = 0
- cursor_state[stage_key] = {"term_index": term_index, "attempt_count": attempts}
-
-
-def _candidate_utility(
- *,
- prev_metrics: Dict[str, Any],
- next_metrics: Dict[str, Any],
- goal_type: str,
- goal_label: str,
- bert_phrase_delta: float,
- chunk_goal_delta: float,
- local_chunk_improved: bool,
-) -> float:
- score_delta = _safe_delta(prev_metrics, next_metrics, "score")
- bm25_delta = _safe_delta(prev_metrics, next_metrics, "bm25_remove_count")
- bert_low_delta = _safe_delta(prev_metrics, next_metrics, "bert_low_count")
- ngram_delta = _safe_delta(prev_metrics, next_metrics, "ngram_signal_count")
- sem_gap_delta = _safe_delta(prev_metrics, next_metrics, "semantic_gap_count")
- title_delta = _safe_delta(prev_metrics, next_metrics, "title_bert_score")
-
- # Dynamic emphasis:
- # - if target phrase is still far from threshold, prioritize phrase-level BERT gains
- # - but keep non-BERT regressions as penalties to preserve future optimization capacity
- key = (goal_label or "").strip().lower()
- prev_phrase = float((prev_metrics.get("bert_phrase_scores") or {}).get(key, 0.0))
- bert_push_mode = (goal_type == "bert" and prev_phrase < BERT_TARGET_THRESHOLD)
-
- w_phrase = 7.5 if bert_push_mode else 3.0
- w_chunk = 1.6 if bert_push_mode else 1.0
- w_score = 1.0
-
- utility = (
- (w_phrase * float(bert_phrase_delta))
- + (w_chunk * float(chunk_goal_delta))
- + (w_score * float(score_delta))
- )
- if local_chunk_improved:
- utility += 0.3
-
- # Cross-metric guardrails as soft penalties (in addition to hard validity checks).
- utility -= max(0.0, bm25_delta) * 1.8
- utility -= max(0.0, bert_low_delta) * 2.4
- utility -= max(0.0, ngram_delta) * 0.6
- utility -= max(0.0, sem_gap_delta) * 1.5
- utility += min(0.0, title_delta) * 1.2
- if goal_type == "title":
- utility += max(0.0, title_delta) * 5.0
- return round(float(utility), 4)
-
-
-def _is_candidate_valid(
- prev_metrics: Dict[str, Any],
- next_metrics: Dict[str, Any],
- goal_type: str,
- goal_label: str,
- optimization_mode: str,
-) -> Tuple[bool, List[str], bool]:
- mode = (optimization_mode or "balanced").lower()
- if mode not in {"conservative", "balanced", "aggressive"}:
- mode = "balanced"
-
- cfg = {
- "conservative": {"max_score_drop": 0.0, "max_title_drop": 0.02},
- "balanced": {"max_score_drop": 1.0, "max_title_drop": 0.03},
- "aggressive": {"max_score_drop": 2.0, "max_title_drop": 0.05},
- }[mode]
-
- reasons = []
- score_drop = float(prev_metrics["score"]) - float(next_metrics["score"])
- if score_drop > cfg["max_score_drop"]:
- reasons.append(f"score_drop>{cfg['max_score_drop']}")
-
- # Hard regressions in critical counters.
- if next_metrics["bm25_remove_count"] > prev_metrics["bm25_remove_count"] + (1 if mode == "aggressive" else 0):
- reasons.append("bm25_remove_regression")
- if next_metrics["bert_low_count"] > prev_metrics["bert_low_count"] + (1 if mode == "aggressive" else 0):
- reasons.append("bert_low_regression")
- if next_metrics["semantic_gap_count"] > prev_metrics["semantic_gap_count"] + (1 if mode == "aggressive" else 0):
- reasons.append("semantic_gap_regression")
-
- prev_title = prev_metrics.get("title_bert_score")
- next_title = next_metrics.get("title_bert_score")
- if goal_type != "title":
- if prev_title is not None and next_title is not None and next_title < (prev_title - cfg["max_title_drop"]):
- reasons.append("title_bert_drop")
-
- improved = _goal_improved(goal_type, goal_label, prev_metrics, next_metrics)
-
- # In conservative mode require explicit goal improvement.
- if mode == "conservative" and not improved:
- reasons.append("goal_not_improved")
-
- return (len(reasons) == 0), reasons, improved
-
-
-def optimize_text(
- request_data: Dict[str, Any],
- progress_callback: Optional[Any] = None,
- cancel_event: Optional[Any] = None,
-) -> Dict[str, Any]:
- target_text = str(request_data.get("target_text", "")).strip()
- competitors = [str(x) for x in (request_data.get("competitors") or []) if str(x).strip()]
- keywords = [str(x) for x in (request_data.get("keywords") or []) if str(x).strip()]
- language = str(request_data.get("language", "en")).strip() or "en"
- target_title = str(request_data.get("target_title", "") or "")
- competitor_titles = [str(x) for x in (request_data.get("competitor_titles") or [])]
-
- diff_mode_used = str(request_data.get("diff_mode", "diff_from_input") or "diff_from_input").strip().lower()
- if diff_mode_used not in {"diff_from_input", "diff_from_original"}:
- diff_mode_used = "diff_from_input"
- diff_base_text = target_text
- diff_base_title = target_title
- if diff_mode_used == "diff_from_original":
- diff_base_text = str(request_data.get("original_target_text") or target_text or "").strip()
- diff_base_title = str(request_data.get("original_target_title") or target_title or "").strip()
- else:
- diff_base_text = (target_text or "").strip()
- diff_base_title = (target_title or "").strip()
-
- api_key = str(request_data.get("api_key", "")).strip()
- if not api_key:
- raise ValueError("API key is required.")
- base_url = str(request_data.get("api_base_url", "https://api.deepseek.com/v1")).strip() or "https://api.deepseek.com/v1"
- model = str(request_data.get("model", "deepseek-chat")).strip() or "deepseek-chat"
- max_iterations = int(request_data.get("max_iterations", 2) or 2)
- max_iterations = max(1, min(8, max_iterations))
- candidates_per_iteration = int(request_data.get("candidates_per_iteration", 2) or 2)
- candidates_per_iteration = max(1, min(5, candidates_per_iteration))
- temperature = float(request_data.get("temperature", 0.25) or 0.25)
- optimization_mode = str(request_data.get("optimization_mode", "balanced") or "balanced")
- phrase_strategy_mode = str(request_data.get("phrase_strategy_mode", "auto") or "auto").strip().lower()
- if phrase_strategy_mode not in {"auto", "exact_preferred", "distributed_preferred", "ensemble"}:
- phrase_strategy_mode = "auto"
- bert_stage_target = float(request_data.get("bert_stage_target", BERT_TARGET_THRESHOLD) or BERT_TARGET_THRESHOLD)
- bert_stage_target = max(0.0, min(1.0, bert_stage_target))
- stage_goal_overrides = request_data.get("stage_goal_overrides") or {}
- if not isinstance(stage_goal_overrides, dict):
- stage_goal_overrides = {}
- req_enabled_stages = request_data.get("enabled_stages") or []
- active_stage_order: List[str] = []
- if isinstance(req_enabled_stages, list):
- for x in req_enabled_stages:
- st = _normalize_stage_name(x)
- if st and st not in active_stage_order:
- active_stage_order.append(st)
- if not active_stage_order:
- active_stage_order = list(STAGE_ORDER)
-
- baseline_analysis = _build_analysis_snapshot(
- target_text, competitors, keywords, language, target_title, competitor_titles
- )
- baseline_semantic = _build_semantic_snapshot(target_text, competitors, language)
- baseline_metrics = _compute_metrics(
- baseline_analysis, baseline_semantic, keywords, language, bert_stage_target=bert_stage_target
- )
-
- # Per-goal iteration budget scales with deficit; total loop steps = sum(effective iters per goal).
- baseline_goal_counts = {
- st: len(
- _collect_optimization_goals(
- baseline_analysis,
- baseline_semantic,
- keywords,
- language,
- stage=st,
- bert_stage_target=bert_stage_target,
- stage_goal_overrides=stage_goal_overrides,
- )
- )
- for st in active_stage_order
- }
- ngram_row_count = int(baseline_goal_counts.get("ngram", 0))
- total_loop_steps = _estimate_total_loop_budget(
- baseline_analysis,
- baseline_semantic,
- keywords,
- language,
- max_iterations,
- candidates_per_iteration,
- bert_stage_target,
- active_stage_order=active_stage_order,
- stage_goal_overrides=stage_goal_overrides,
- )
-
- current_text = target_text
- current_title = (target_title or "").strip()
- current_analysis = baseline_analysis
- current_semantic = baseline_semantic
- current_metrics = baseline_metrics
- logs: List[Dict[str, Any]] = []
- applied_changes = 0
-
- def _emit(ev: str, **kwargs: Any) -> None:
- if progress_callback:
- try:
- progress_callback({"event": ev, **kwargs})
- except Exception:
- pass
-
- def _cancelled() -> bool:
- return cancel_event is not None and getattr(cancel_event, "is_set", lambda: False)()
-
- candidate_eval_cache: Dict[Tuple[str, str], Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any]]] = {}
-
- def _evaluate_candidate_state(body_text: str, title_text: str) -> Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any]]:
- key = ((body_text or "").strip(), (title_text or "").strip())
- cached = candidate_eval_cache.get(key)
- if cached is not None:
- return cached
- analysis = _build_analysis_snapshot(key[0], competitors, keywords, language, key[1], competitor_titles)
- semantic = _build_semantic_snapshot(key[0], competitors, language)
- metrics = _compute_metrics(analysis, semantic, keywords, language, bert_stage_target=bert_stage_target)
- candidate_eval_cache[key] = (analysis, semantic, metrics)
- return analysis, semantic, metrics
-
- def _pack_result(stopped_early: bool = False, stop_reason: str = "") -> Dict[str, Any]:
- # Title string must match what last metrics used (title_analysis.target_title), not only the mutable var.
- _ta = (current_analysis or {}).get("title_analysis") or {}
- _ot = ""
- if isinstance(_ta, dict) and (_ta.get("target_title") or "").strip():
- _ot = str(_ta.get("target_title")).strip()
- if not _ot:
- _ot = (current_title or "").strip()
- if not _ot:
- _ot = (target_title or "").strip()
-
- diff_body_html = ""
- diff_changes: List[Dict[str, str]] = []
- if (diff_base_text or "").strip() != (current_text or "").strip():
- dh, dc = _diff_sentences_html(diff_base_text or "", current_text or "")
- if dc:
- diff_body_html = dh
- diff_changes = dc
-
- diff_title_html = ""
- diff_title_changes: List[Dict[str, str]] = []
- if (diff_base_title or "").strip() != (_ot or "").strip():
- dth, dtc = _diff_title_html(diff_base_title or "", _ot or "")
- if dtc:
- diff_title_html = dth
- diff_title_changes = dtc
- return {
- "ok": True,
- "optimized_text": current_text,
- "optimized_title": _ot,
- "baseline_metrics": baseline_metrics,
- "final_metrics": current_metrics,
- "iterations": logs,
- "applied_changes": applied_changes,
- "optimization_mode": optimization_mode,
- "phrase_strategy_mode": phrase_strategy_mode,
- "bert_stage_target": round(bert_stage_target, 4),
- "diff_mode": diff_mode_used,
- "diff_body_html": diff_body_html,
- "diff_title_html": diff_title_html,
- "diff_changes": diff_changes,
- "diff_title_changes": diff_title_changes,
- "stopped_early": stopped_early,
- "stop_reason": stop_reason,
- }
-
- _emit("preparing", message="Подготовка", phase="baseline")
- _emit(
- "started",
- total_steps=total_loop_steps,
- max_iterations_setting=max_iterations,
- ngram_targets=ngram_row_count,
- stages_order=list(active_stage_order),
- )
-
- seen_candidate_rewrites = set()
- cascade_level = 1
- consecutive_failures = 0
- goal_attempt_cursor: Dict[str, int] = {}
- attempted_spans = set()
- queued_candidates: List[Dict[str, Any]] = []
- stage_idx = 0
- stage_no_progress_steps = 0
- stage_goal_cursor: Dict[str, Dict[str, int]] = {}
-
- for step in range(total_loop_steps):
- if _cancelled():
- logs.append(
- {
- "step": step + 1,
- "status": "stopped",
- "reason": "Остановка пользователем (сохранён текущий текст и метрики).",
- }
- )
- return _pack_result(stopped_early=True, stop_reason="user_cancelled")
-
- while stage_idx < len(active_stage_order) and _is_stage_complete(
- active_stage_order[stage_idx], current_metrics, bert_stage_target=bert_stage_target
- ):
- stage_idx += 1
- stage_no_progress_steps = 0
- if stage_idx >= len(active_stage_order):
- logs.append({"step": step + 1, "status": "stopped", "reason": "All optimization stages completed."})
- break
-
- active_stage = active_stage_order[stage_idx]
- goals_for_stage = _collect_optimization_goals(
- current_analysis,
- current_semantic,
- keywords,
- language,
- stage=active_stage,
- bert_stage_target=bert_stage_target,
- stage_goal_overrides=stage_goal_overrides,
- )
- state = stage_goal_cursor.get(active_stage) or {"goal_index": 0, "attempt_count": 0}
- goal_index = int(state.get("goal_index", 0))
- attempt_count = int(state.get("attempt_count", 0))
-
- # Advance across goals that exhausted per-goal iteration budget (scaled by deficit).
- while goal_index < len(goals_for_stage):
- g_try = goals_for_stage[goal_index]
- eff_max_iter, _ = _per_goal_budget(g_try, max_iterations, candidates_per_iteration, bert_stage_target)
- if attempt_count < eff_max_iter:
- break
- goal_index += 1
- attempt_count = 0
-
- if goal_index >= len(goals_for_stage):
- stage_idx += 1
- stage_no_progress_steps = 0
- logs.append(
- {
- "step": step + 1,
- "status": "stage_skipped",
- "stage": active_stage,
- "reason": f"All goals exhausted for stage '{active_stage}' (per-goal iteration budget).",
- }
- )
- stage_goal_cursor[active_stage] = {"goal_index": goal_index, "attempt_count": attempt_count}
- continue
-
- goal = goals_for_stage[goal_index]
- eff_max_iter, eff_cand = _per_goal_budget(goal, max_iterations, candidates_per_iteration, bert_stage_target)
- attempt_count += 1
- stage_goal_cursor[active_stage] = {"goal_index": goal_index, "attempt_count": attempt_count}
- if goal["type"] == "none":
- stage_idx += 1
- stage_no_progress_steps = 0
- logs.append(
- {
- "step": step + 1,
- "status": "stage_skipped",
- "stage": active_stage,
- "reason": f"No actionable goals for stage '{active_stage}', moving to next stage.",
- }
- )
- continue
-
- _emit(
- "step_start",
- step=step + 1,
- total_steps=total_loop_steps,
- active_stage=active_stage,
- goal_type=goal.get("type"),
- goal_label=goal.get("label"),
- score=current_metrics.get("score"),
- goal_budget_iter=eff_max_iter,
- goal_budget_candidates=eff_cand,
- )
-
- goal_key = f"{goal.get('type', '')}:{goal.get('label', '')}".strip().lower()
- base_attempt_cursor = int(goal_attempt_cursor.get(goal_key, 0))
- candidates: List[Dict[str, Any]] = []
- chosen_spans: List[Dict[str, Any]] = []
- candidate_idx = 0
- span_trials_eff = 1
-
- if goal.get("type") == "title":
- if not current_title:
- logs.append(
- {
- "step": step + 1,
- "status": "stage_skipped",
- "stage": active_stage,
- "reason": "Поле Title пустое — этап title пропущен.",
- }
- )
- stage_idx += 1
- continue
-
- chosen_spans = [
- {
- "operation": "title_rewrite",
- "span_start": 0,
- "span_end": 0,
- "sentence_index": 0,
- "span_variant": 0,
- "sentence_before": current_title,
- }
- ]
- original_span_text = current_title
- body_excerpt = (current_text or "").strip()[:1200]
- comp_ctx = " | ".join(t.strip() for t in competitor_titles[:4] if t.strip())
- competitor_title_hint = (
- f"Примеры title конкурентов (только стиль/длина): {comp_ctx}" if comp_ctx else ""
- )
- strategy_plan = _build_phrase_strategy_plan(
- phrase_strategy_mode,
- "title",
- str(goal.get("label", "")),
- eff_cand,
- )
- for strategy_variant in strategy_plan:
- candidate_idx += 1
- temp = min(1.1, max(0.0, temperature + (candidate_idx - 1) * 0.07))
- if _cancelled():
- logs.append(
- {
- "step": step + 1,
- "status": "stopped",
- "reason": "Остановка пользователем перед запросом к LLM.",
- }
- )
- return _pack_result(stopped_early=True, stop_reason="user_cancelled")
- _emit(
- "llm_call",
- candidate_index=candidate_idx,
- span_trial=1,
- span_trials=1,
- strategy=strategy_variant,
- )
- try:
- llm_result = _llm_edit_title(
- api_key=api_key,
- base_url=base_url,
- model=model,
- language=language,
- current_title=current_title,
- body_excerpt=body_excerpt,
- competitor_title_hint=competitor_title_hint,
- focus_terms=goal.get("focus_terms", []) or [],
- avoid_terms=goal.get("avoid_terms", []) or [],
- keywords=keywords,
- cascade_level=cascade_level,
- temperature=temp,
- phrase_strategy_mode=strategy_variant,
- )
- edited_text = str((llm_result or {}).get("edited_text", "")).strip()
- llm_rationale = str((llm_result or {}).get("rationale", "")).strip()
- prompt_debug = (llm_result or {}).get("prompt_debug", {})
- if not edited_text or edited_text == original_span_text.strip():
- continue
-
- quality_issues = _validate_title_candidate(edited_text)
- cand_analysis, cand_semantic, cand_metrics = _evaluate_candidate_state(current_text, edited_text)
- before_rel = float(current_metrics.get("title_bert_score") or 0.0)
- after_rel = float(cand_metrics.get("title_bert_score") or 0.0)
- chunk_delta = round(after_rel - before_rel, 4)
- local_chunk_improved = chunk_delta >= _min_chunk_delta("title")
- bert_phrase_delta = 0.0
- valid, invalid_reasons, goal_improved = _is_candidate_valid(
- current_metrics, cand_metrics, goal["type"], goal["label"], optimization_mode
- )
- delta_score = round(cand_metrics["score"] - current_metrics["score"], 3)
- candidate_utility = _candidate_utility(
- prev_metrics=current_metrics,
- next_metrics=cand_metrics,
- goal_type=str(goal.get("type", "")),
- goal_label=str(goal.get("label", "")),
- bert_phrase_delta=bert_phrase_delta,
- chunk_goal_delta=chunk_delta,
- local_chunk_improved=local_chunk_improved,
- )
- md = _metrics_delta(current_metrics, cand_metrics)
-
- if quality_issues:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": "quality_validation_failed",
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "invalid_reasons": quality_issues,
- "delta_score": -999.0,
- "candidate_score": None,
- "sentence_after": edited_text,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "operation": "title_rewrite",
- "sentence_index": 0,
- "span_start": 0,
- "span_end": 0,
- "span_variant": 0,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": original_span_text,
- }
- )
- continue
-
- candidate_key = ("title_rewrite", 0, 0, edited_text.strip().lower())
- if candidate_key in seen_candidate_rewrites:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": "duplicate_candidate_rewrite",
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "invalid_reasons": ["duplicate_candidate_rewrite"],
- "delta_score": -999.0,
- "candidate_score": None,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "operation": "title_rewrite",
- "sentence_index": 0,
- "span_start": 0,
- "span_end": 0,
- "span_variant": 0,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": original_span_text,
- }
- )
- continue
- seen_candidate_rewrites.add(candidate_key)
-
- candidate_text = current_text
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "sentence_before": original_span_text,
- "sentence_after": edited_text,
- "operation": "title_rewrite",
- "sentence_index": 0,
- "span_start": 0,
- "span_end": 0,
- "span_variant": 0,
- "phrase_strategy_used": strategy_variant,
- "text": candidate_text,
- "new_title": edited_text,
- "analysis": cand_analysis,
- "semantic": cand_semantic,
- "metrics": cand_metrics,
- "valid": valid,
- "goal_improved": goal_improved,
- "bert_phrase_delta": bert_phrase_delta,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "invalid_reasons": invalid_reasons,
- "delta_score": delta_score,
- "candidate_score": cand_metrics.get("score"),
- "candidate_utility": candidate_utility,
- "metrics_delta": md,
- "edit_payload": {
- "operation": "title_rewrite",
- "span_start": 0,
- "span_end": 0,
- "edited_text": edited_text,
- },
- }
- )
- except Exception as e:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": str(e),
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": False,
- "chunk_goal_delta": -999.0,
- "invalid_reasons": [str(e)],
- "delta_score": -999.0,
- "candidate_score": None,
- "llm_prompt_debug": {
- "operation": "title_rewrite",
- "cascade_level": cascade_level,
- "goal_type": goal.get("type"),
- "goal_label": goal.get("label"),
- "phrase_strategy_mode": strategy_variant,
- },
- "llm_rationale": "",
- "operation": "title_rewrite",
- "sentence_index": 0,
- "span_start": 0,
- "span_end": 0,
- "span_variant": 0,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": current_title,
- }
- )
-
- else:
- sentences = _split_sentences(current_text)
- if not sentences:
- logs.append({"step": step + 1, "status": "stopped", "reason": "No sentences available for editing."})
- break
-
- span_trials = _portfolio_span_trials(str(goal.get("type", "")), cascade_level, eff_cand)
- local_candidates = max(eff_cand, span_trials)
- if cascade_level > 2:
- local_candidates = min(6, max(local_candidates, eff_cand + 1))
- span_trials_eff = span_trials
-
- for st in range(span_trials):
- attempt_cursor = base_attempt_cursor + st
- operation, span_start, span_end, sent_idx, span_variant = _choose_edit_span(
- sentences, goal, language, cascade_level, attempt_cursor
- )
- max_span_retries = max(1, len(sentences) * 4)
- retries = 0
- while retries < max_span_retries:
- span_key = (goal_key, cascade_level, operation, span_start, span_end)
- if span_key not in attempted_spans:
- attempted_spans.add(span_key)
- break
- attempt_cursor += 1
- operation, span_start, span_end, sent_idx, span_variant = _choose_edit_span(
- sentences, goal, language, cascade_level, attempt_cursor
- )
- retries += 1
-
- original_span_text = " ".join(sentences[span_start : span_end + 1]).strip()
- context_before = " ".join(sentences[max(0, span_start - 2) : span_start]).strip()
- context_after = " ".join(sentences[span_end + 1 : min(len(sentences), span_end + 3)]).strip()
- chosen_spans.append(
- {
- "operation": operation,
- "span_start": span_start,
- "span_end": span_end,
- "sentence_index": sent_idx,
- "span_variant": span_variant,
- "sentence_before": original_span_text,
- }
- )
-
- per_span_candidates = max(1, local_candidates // span_trials)
- strategy_plan = _build_phrase_strategy_plan(
- phrase_strategy_mode,
- str(goal.get("type", "")),
- str(goal.get("label", "")),
- per_span_candidates,
- )
- for strategy_variant in strategy_plan:
- candidate_idx += 1
- temp = min(1.1, max(0.0, temperature + (candidate_idx - 1) * 0.07))
- if _cancelled():
- logs.append(
- {
- "step": step + 1,
- "status": "stopped",
- "reason": "Остановка пользователем перед запросом к LLM.",
- }
- )
- return _pack_result(stopped_early=True, stop_reason="user_cancelled")
- _emit(
- "llm_call",
- candidate_index=candidate_idx,
- span_trial=st + 1,
- span_trials=span_trials,
- strategy=strategy_variant,
- )
- try:
- llm_result = _llm_edit_chunk(
- api_key=api_key,
- base_url=base_url,
- model=model,
- language=language,
- full_text=current_text,
- chunk_text=original_span_text,
- operation=operation,
- context_before=context_before,
- context_after=context_after,
- cascade_level=cascade_level,
- goal_type=goal["type"],
- goal_label=goal["label"],
- focus_terms=goal["focus_terms"],
- avoid_terms=goal["avoid_terms"],
- temperature=temp,
- phrase_strategy_mode=strategy_variant,
- )
- edited_text = str((llm_result or {}).get("edited_text", "")).strip()
- llm_rationale = str((llm_result or {}).get("rationale", "")).strip()
- prompt_debug = (llm_result or {}).get("prompt_debug", {})
- if not edited_text or edited_text == original_span_text:
- continue
-
- quality_issues = _validate_candidate_text(
- edited_text,
- cascade_level,
- operation,
- goal_label=goal.get("label", ""),
- focus_terms=goal.get("focus_terms", []) or [],
- )
- before_rel, after_rel = _chunk_relevance_pair(
- original_span_text,
- edited_text,
- goal["type"],
- goal["label"],
- goal.get("focus_terms", []) or [],
- language,
- goal,
- )
- chunk_delta = _chunk_goal_delta(
- original_span_text,
- edited_text,
- goal["type"],
- goal["label"],
- goal.get("focus_terms", []) or [],
- language,
- goal,
- )
- local_chunk_improved = chunk_delta >= _min_chunk_delta(goal["type"])
-
- if quality_issues:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": "quality_validation_failed",
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "invalid_reasons": quality_issues,
- "delta_score": -999.0,
- "candidate_score": None,
- "sentence_after": edited_text,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "operation": operation,
- "sentence_index": sent_idx,
- "span_start": span_start,
- "span_end": span_end,
- "span_variant": span_variant,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": original_span_text,
- }
- )
- continue
-
- prefilter_reasons = _local_prefilter_reasons(
- goal_type=str(goal.get("type", "")),
- focus_terms=goal.get("focus_terms", []) or [],
- edited_text=edited_text,
- chunk_delta=chunk_delta,
- before_rel=before_rel,
- after_rel=after_rel,
- )
- if prefilter_reasons:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": "local_prefilter_rejected",
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "invalid_reasons": prefilter_reasons,
- "delta_score": -999.0,
- "candidate_score": None,
- "sentence_after": edited_text,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "operation": operation,
- "sentence_index": sent_idx,
- "span_start": span_start,
- "span_end": span_end,
- "span_variant": span_variant,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": original_span_text,
- }
- )
- continue
-
- candidate_key = (operation, span_start, span_end, edited_text.strip().lower())
- if candidate_key in seen_candidate_rewrites:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": "duplicate_candidate_rewrite",
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "invalid_reasons": ["duplicate_candidate_rewrite"],
- "delta_score": -999.0,
- "candidate_score": None,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "operation": operation,
- "sentence_index": sent_idx,
- "span_start": span_start,
- "span_end": span_end,
- "span_variant": span_variant,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": original_span_text,
- }
- )
- continue
- seen_candidate_rewrites.add(candidate_key)
-
- if operation == "insert":
- candidate_sentences = _insert_after(sentences, span_end, edited_text)
- else:
- candidate_sentences = _replace_span(sentences, span_start, span_end, edited_text)
- candidate_text = " ".join(candidate_sentences).strip()
-
- cand_analysis, cand_semantic, cand_metrics = _evaluate_candidate_state(candidate_text, current_title)
- valid, invalid_reasons, goal_improved = _is_candidate_valid(
- current_metrics, cand_metrics, goal["type"], goal["label"], optimization_mode
- )
- delta_score = round(cand_metrics["score"] - current_metrics["score"], 3)
- bert_phrase_delta = _bert_phrase_delta(goal["label"], current_metrics, cand_metrics) if goal.get("type") == "bert" else 0.0
- candidate_utility = _candidate_utility(
- prev_metrics=current_metrics,
- next_metrics=cand_metrics,
- goal_type=str(goal.get("type", "")),
- goal_label=str(goal.get("label", "")),
- bert_phrase_delta=bert_phrase_delta,
- chunk_goal_delta=chunk_delta,
- local_chunk_improved=local_chunk_improved,
- )
- md = _metrics_delta(current_metrics, cand_metrics)
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "sentence_before": original_span_text,
- "sentence_after": edited_text,
- "operation": operation,
- "sentence_index": sent_idx,
- "span_start": span_start,
- "span_end": span_end,
- "span_variant": span_variant,
- "phrase_strategy_used": strategy_variant,
- "text": candidate_text,
- "analysis": cand_analysis,
- "semantic": cand_semantic,
- "metrics": cand_metrics,
- "valid": valid,
- "goal_improved": goal_improved,
- "bert_phrase_delta": bert_phrase_delta,
- "local_chunk_improved": local_chunk_improved,
- "chunk_goal_delta": chunk_delta,
- "chunk_relevance_before": before_rel,
- "chunk_relevance_after": after_rel,
- "term_diff": _term_diff(original_span_text, edited_text, language),
- "llm_prompt_debug": prompt_debug,
- "llm_rationale": llm_rationale,
- "invalid_reasons": invalid_reasons,
- "delta_score": delta_score,
- "candidate_score": cand_metrics.get("score"),
- "candidate_utility": candidate_utility,
- "metrics_delta": md,
- "edit_payload": {
- "operation": operation,
- "span_start": span_start,
- "span_end": span_end,
- "edited_text": edited_text,
- },
- }
- )
- except Exception as e:
- candidates.append(
- {
- "candidate_index": candidate_idx,
- "error": str(e),
- "valid": False,
- "goal_improved": False,
- "local_chunk_improved": False,
- "chunk_goal_delta": -999.0,
- "invalid_reasons": [str(e)],
- "delta_score": -999.0,
- "candidate_score": None,
- "llm_prompt_debug": {
- "operation": operation,
- "cascade_level": cascade_level,
- "goal_type": goal.get("type"),
- "goal_label": goal.get("label"),
- "phrase_strategy_mode": strategy_variant,
- },
- "llm_rationale": "",
- "operation": operation,
- "sentence_index": sent_idx,
- "span_start": span_start,
- "span_end": span_end,
- "span_variant": span_variant,
- "phrase_strategy_used": strategy_variant,
- "sentence_before": original_span_text,
- }
- )
-
- goal_attempt_cursor[goal_key] = base_attempt_cursor + span_trials_eff
- primary_span = chosen_spans[0] if chosen_spans else {"operation": "-", "span_start": 0, "span_end": 0, "sentence_index": 0, "span_variant": 0, "sentence_before": ""}
-
- valid_raw_candidates = [c for c in candidates if c.get("valid")]
- valid_candidates = [
- c
- for c in valid_raw_candidates
- if c.get("valid")
- and (
- c.get("goal_improved")
- or (goal.get("type") == "bert" and float(c.get("bert_phrase_delta") or 0.0) > 0.0)
- or float(c.get("candidate_score") or -1) > float(current_metrics["score"])
- or float(c.get("candidate_utility") or -999.0) > 0.0
- )
- and (
- goal.get("type") != "bert"
- or float(c.get("bert_phrase_delta") or 0.0) > 0.0
- or c.get("local_chunk_improved")
- )
- ]
- if not valid_candidates:
- # Local-first accumulation mode:
- # if we have guardrail-valid candidates that improve chunk relevance,
- # apply the strongest local edit immediately and continue optimizing next chunks.
- local_progress_candidates = [
- c
- for c in valid_raw_candidates
- if c.get("local_chunk_improved")
- ]
- if local_progress_candidates:
- best_local = sorted(
- local_progress_candidates,
- key=lambda c: (
- float(c.get("candidate_utility") or -999.0),
- float(c.get("chunk_goal_delta") or 0.0),
- float(c.get("bert_phrase_delta") or 0.0),
- float(c.get("candidate_score") or -999.0),
- ),
- reverse=True,
- )[0]
-
- prev_metrics = current_metrics
- current_text = best_local["text"]
- if str(best_local.get("operation") or "") == "title_rewrite":
- nt = (best_local.get("new_title") or best_local.get("sentence_after") or "").strip()
- if nt:
- current_title = nt
- current_analysis = best_local["analysis"]
- current_semantic = best_local["semantic"]
- current_metrics = best_local["metrics"]
- progressed_stage = _stage_primary_progress(active_stage, prev_metrics, current_metrics)
- if progressed_stage:
- stage_no_progress_steps = 0
- else:
- stage_no_progress_steps += 1
- applied_changes += 1
- queued_candidates = []
-
- logs.append(
- {
- "step": step + 1,
- "status": "applied_local_progress",
- "stage": active_stage,
- "goal": goal,
- "cascade_level": cascade_level,
- "operation": best_local.get("operation"),
- "sentence_index": best_local.get("sentence_index"),
- "span_start": best_local.get("span_start"),
- "span_end": best_local.get("span_end"),
- "span_variant": best_local.get("span_variant"),
- "sentence_before": best_local.get("sentence_before"),
- "sentence_after": best_local.get("sentence_after"),
- "reason": "Applied best local-improvement candidate despite no immediate global gain.",
- "current_score": prev_metrics.get("score"),
- "metrics_before": prev_metrics,
- "metrics_after": current_metrics,
- "delta_score": round(float(current_metrics.get("score", 0)) - float(prev_metrics.get("score", 0)), 3),
- "chosen_candidate_index": best_local.get("candidate_index"),
- "chosen_chunk_goal_delta": best_local.get("chunk_goal_delta"),
- "chosen_bert_phrase_delta": best_local.get("bert_phrase_delta"),
- "chosen_candidate_utility": best_local.get("candidate_utility"),
- "chosen_metrics_delta": best_local.get("metrics_delta"),
- "candidates": [
- {
- "candidate_index": c.get("candidate_index"),
- "valid": c.get("valid", False),
- "goal_improved": c.get("goal_improved", False),
- "bert_phrase_delta": c.get("bert_phrase_delta"),
- "local_chunk_improved": c.get("local_chunk_improved", False),
- "chunk_goal_delta": c.get("chunk_goal_delta"),
- "chunk_relevance_before": c.get("chunk_relevance_before"),
- "chunk_relevance_after": c.get("chunk_relevance_after"),
- "term_diff": c.get("term_diff"),
- "llm_prompt_debug": c.get("llm_prompt_debug"),
- "llm_rationale": c.get("llm_rationale"),
- "metrics_delta": c.get("metrics_delta"),
- "candidate_utility": c.get("candidate_utility"),
- "invalid_reasons": c.get("invalid_reasons", []),
- "delta_score": c.get("delta_score"),
- "candidate_score": c.get("candidate_score"),
- "sentence_after": c.get("sentence_after"),
- "error": c.get("error"),
- }
- for c in candidates
- ],
- }
- )
- consecutive_failures = 0
- cascade_level = 1
- goal_attempt_cursor[goal_key] = base_attempt_cursor + 1
- continue
-
- local_pool = [
- c
- for c in candidates
- if c.get("local_chunk_improved")
- and c.get("edit_payload")
- and c.get("candidate_score") is not None
- and goal.get("type") != "title"
- ]
- local_pool.sort(
- key=lambda c: (
- float(c.get("candidate_utility") or -999.0),
- float(c.get("chunk_goal_delta") or -999.0),
- float(c.get("candidate_score") or -999.0),
- ),
- reverse=True,
- )
- for c in local_pool[:4]:
- queue_key = (
- goal_key,
- c.get("operation"),
- c.get("span_start"),
- c.get("span_end"),
- str((c.get("sentence_after") or "")).strip().lower(),
- )
- if not any(x.get("queue_key") == queue_key for x in queued_candidates):
- queued_candidates.append({"queue_key": queue_key, "candidate": c})
-
- batch_applied = False
- batch_info: Dict[str, Any] = {}
- if len(queued_candidates) >= 2 and goal.get("type") != "title":
- pool = [x["candidate"] for x in queued_candidates[:6]]
- combos = _non_conflicting_edit_combos(pool, min_size=2, max_size=4)
- best_batch: Optional[Dict[str, Any]] = None
- prev_metrics = current_metrics
- for combo in combos:
- edits = [c.get("edit_payload") for c in combo if c.get("edit_payload")]
- if len(edits) != len(combo):
- continue
- batch_sentences = _apply_edits_to_sentences(sentences, edits)
- batch_text = " ".join(batch_sentences).strip()
- batch_analysis, batch_semantic, batch_metrics = _evaluate_candidate_state(batch_text, current_title)
- b_valid, b_reasons, b_goal = _is_candidate_valid(
- current_metrics, batch_metrics, goal["type"], goal["label"], optimization_mode
- )
- b_delta = round(batch_metrics["score"] - current_metrics["score"], 3)
- local_sum = sum(float(c.get("chunk_goal_delta") or 0.0) for c in combo)
- if not (b_valid and (b_goal or b_delta > 0)):
- continue
- if goal.get("type") == "bert" and local_sum < (_min_chunk_delta("bert") * len(combo)):
- continue
- cand = {
- "combo": combo,
- "batch_text": batch_text,
- "batch_analysis": batch_analysis,
- "batch_semantic": batch_semantic,
- "batch_metrics": batch_metrics,
- "b_delta": b_delta,
- "local_sum": local_sum,
- "b_reasons": b_reasons,
- "b_goal": b_goal,
- }
- if best_batch is None or (
- cand["b_delta"],
- cand["local_sum"],
- len(cand["combo"]),
- ) > (
- best_batch["b_delta"],
- best_batch["local_sum"],
- len(best_batch["combo"]),
- ):
- best_batch = cand
- if best_batch:
- current_text = best_batch["batch_text"]
- current_analysis = best_batch["batch_analysis"]
- current_semantic = best_batch["batch_semantic"]
- current_metrics = best_batch["batch_metrics"]
- progressed_stage = _stage_primary_progress(active_stage, prev_metrics, current_metrics)
- if progressed_stage:
- stage_no_progress_steps = 0
- else:
- stage_no_progress_steps += 1
- applied_changes += 1
- batch_applied = True
- batch_info = {
- "status": "applied_batch",
- "batch_candidate_ids": [c.get("candidate_index") for c in best_batch["combo"]],
- "batch_size": len(best_batch["combo"]),
- "batch_local_chunk_delta_sum": round(best_batch["local_sum"], 4),
- "delta_score": best_batch["b_delta"],
- "metrics_before": prev_metrics,
- "metrics_after": best_batch["batch_metrics"],
- "metrics_delta": _metrics_delta(prev_metrics, best_batch["batch_metrics"]),
- }
- queued_candidates = []
- consecutive_failures = 0
- cascade_level = 1
- goal_attempt_cursor[goal_key] = 0
- if not batch_applied:
- batch_info = {
- "status": "batch_rejected",
- "reason": "Queued local improvements could not pass global constraints together.",
- }
-
- if batch_applied:
- logs.append(
- {
- "step": step + 1,
- "status": "applied_batch",
- "stage": active_stage,
- "goal": goal,
- "cascade_level": cascade_level,
- "operation": "batch",
- "sentence_index": primary_span.get("sentence_index"),
- "span_start": primary_span.get("span_start"),
- "span_end": primary_span.get("span_end"),
- "span_variant": primary_span.get("span_variant"),
- "sentence_before": primary_span.get("sentence_before"),
- "current_score": (batch_info.get("metrics_before") or {}).get("score"),
- "metrics_before": batch_info.get("metrics_before"),
- "metrics_after": current_metrics,
- "reason": "Applied queued local-improvement edits as a batch.",
- "batch_info": batch_info,
- "candidates": [
- {
- "candidate_index": c.get("candidate_index"),
- "valid": c.get("valid", False),
- "goal_improved": c.get("goal_improved", False),
- "bert_phrase_delta": c.get("bert_phrase_delta"),
- "local_chunk_improved": c.get("local_chunk_improved", False),
- "chunk_goal_delta": c.get("chunk_goal_delta"),
- "chunk_relevance_before": c.get("chunk_relevance_before"),
- "chunk_relevance_after": c.get("chunk_relevance_after"),
- "term_diff": c.get("term_diff"),
- "llm_prompt_debug": c.get("llm_prompt_debug"),
- "llm_rationale": c.get("llm_rationale"),
- "metrics_delta": c.get("metrics_delta"),
- "candidate_utility": c.get("candidate_utility"),
- "invalid_reasons": c.get("invalid_reasons", []),
- "delta_score": c.get("delta_score"),
- "candidate_score": c.get("candidate_score"),
- "sentence_after": c.get("sentence_after"),
- "error": c.get("error"),
- }
- for c in candidates
- ],
- }
- )
- continue
-
- logs.append(
- {
- "step": step + 1,
- "status": "rejected",
- "stage": active_stage,
- "goal": goal,
- "cascade_level": cascade_level,
- "operation": primary_span.get("operation"),
- "sentence_index": primary_span.get("sentence_index"),
- "span_start": primary_span.get("span_start"),
- "span_end": primary_span.get("span_end"),
- "span_variant": primary_span.get("span_variant"),
- "sentence_before": primary_span.get("sentence_before"),
- "current_score": current_metrics["score"],
- "reason": (
- "No valid candidate satisfied constraints."
- if not valid_raw_candidates
- else "Valid candidates existed but none improved goal or total score."
- ),
- "valid_candidates_count": len(valid_raw_candidates),
- "promotable_candidates_count": len(valid_candidates),
- "queued_local_candidates": len(local_pool),
- "queued_total": len(queued_candidates),
- "batch_info": batch_info if batch_info else None,
- "candidates": [
- {
- "candidate_index": c.get("candidate_index"),
- "valid": c.get("valid", False),
- "goal_improved": c.get("goal_improved", False),
- "bert_phrase_delta": c.get("bert_phrase_delta"),
- "local_chunk_improved": c.get("local_chunk_improved", False),
- "chunk_goal_delta": c.get("chunk_goal_delta"),
- "chunk_relevance_before": c.get("chunk_relevance_before"),
- "chunk_relevance_after": c.get("chunk_relevance_after"),
- "term_diff": c.get("term_diff"),
- "llm_prompt_debug": c.get("llm_prompt_debug"),
- "llm_rationale": c.get("llm_rationale"),
- "metrics_delta": c.get("metrics_delta"),
- "candidate_utility": c.get("candidate_utility"),
- "invalid_reasons": c.get("invalid_reasons", []),
- "delta_score": c.get("delta_score"),
- "candidate_score": c.get("candidate_score"),
- "sentence_after": c.get("sentence_after"),
- "error": c.get("error"),
- }
- for c in candidates
- ],
- }
- )
- stage_no_progress_steps += 1
- # Stage transition is controlled by per-stage iteration budget and completion checks.
- consecutive_failures += 1
- if consecutive_failures >= 2 and cascade_level < 4:
- cascade_level += 1
- consecutive_failures = 0
- logs[-1]["escalated_to_level"] = cascade_level
- continue
-
- best = sorted(
- valid_candidates,
- key=lambda c: (
- 1 if c.get("goal_improved") else 0,
- float(c.get("candidate_utility") or -999.0),
- float(c.get("bert_phrase_delta") or 0.0),
- float(c.get("chunk_goal_delta") or 0.0),
- c["metrics"]["score"],
- ),
- reverse=True,
- )[0]
-
- prev_metrics = current_metrics
- current_text = best["text"]
- if str(best.get("operation") or "") == "title_rewrite":
- nt = (best.get("new_title") or best.get("sentence_after") or "").strip()
- if nt:
- current_title = nt
- current_analysis = best["analysis"]
- current_semantic = best["semantic"]
- current_metrics = best["metrics"]
- progressed_stage = _stage_primary_progress(active_stage, prev_metrics, current_metrics)
- if progressed_stage:
- stage_no_progress_steps = 0
- else:
- stage_no_progress_steps += 1
- applied_changes += 1
- queued_candidates = []
-
- logs.append(
- {
- "step": step + 1,
- "status": "applied",
- "stage": active_stage,
- "goal": goal,
- "cascade_level": cascade_level,
- "operation": best.get("operation"),
- "sentence_index": best.get("sentence_index"),
- "span_start": best.get("span_start"),
- "span_end": best.get("span_end"),
- "span_variant": best.get("span_variant"),
- "sentence_before": best.get("sentence_before"),
- "sentence_after": best["sentence_after"],
- "current_score": prev_metrics["score"],
- "metrics_before": prev_metrics,
- "metrics_after": current_metrics,
- "delta_score": round(current_metrics["score"] - prev_metrics["score"], 3),
- "chosen_candidate_index": best.get("candidate_index"),
- "chosen_candidate_utility": best.get("candidate_utility"),
- "candidates": [
- {
- "candidate_index": c.get("candidate_index"),
- "valid": c.get("valid", False),
- "goal_improved": c.get("goal_improved", False),
- "bert_phrase_delta": c.get("bert_phrase_delta"),
- "local_chunk_improved": c.get("local_chunk_improved", False),
- "chunk_goal_delta": c.get("chunk_goal_delta"),
- "chunk_relevance_before": c.get("chunk_relevance_before"),
- "chunk_relevance_after": c.get("chunk_relevance_after"),
- "term_diff": c.get("term_diff"),
- "llm_prompt_debug": c.get("llm_prompt_debug"),
- "llm_rationale": c.get("llm_rationale"),
- "metrics_delta": c.get("metrics_delta"),
- "candidate_utility": c.get("candidate_utility"),
- "invalid_reasons": c.get("invalid_reasons", []),
- "delta_score": c.get("delta_score"),
- "candidate_score": c.get("candidate_score"),
- "sentence_after": c.get("sentence_after"),
- "error": c.get("error"),
- }
- for c in candidates
- ],
- }
- )
- # After successful edit, return to cheapest level and reset failure streak.
- consecutive_failures = 0
- cascade_level = 1
- goal_attempt_cursor[goal_key] = 0
-
- return _pack_result()