ilia
4b034d19f6
Cherry-pick heuristics, manifest merge, global file search, scrollbar styling, and pytest/Playwright coverage unblock review and filtered export for mmetl.
440 lines
13 KiB
Python
440 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Bulk cherry-pick data-rich messages (files, facts) — not conversation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
MANIFEST_PATH = REPO_ROOT / "manifest.json"
|
|
DEFAULT_EXPORT = Path.home() / "Downloads/Levkin Slack export enriched"
|
|
|
|
SKIP_SUBTYPES = {
|
|
"channel_join",
|
|
"channel_leave",
|
|
"channel_topic",
|
|
"channel_purpose",
|
|
"channel_name",
|
|
"channel_archive",
|
|
"channel_unarchive",
|
|
"group_join",
|
|
"group_leave",
|
|
}
|
|
|
|
RETURN_CHANNELS = {"29wilson", "520steeles"}
|
|
|
|
NOISE_TEXT = re.compile(
|
|
r"^(ok|okay|thanks|thx|ty|lol|lmao|\+1|done|yes|no|yep|nope|sure|cool|nice|got it|"
|
|
r"will do|sounds good|perfect|great|👍|👌)\.?!?$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# Concrete facts — not decision/planning chatter that merely mentions a topic word.
|
|
DATA_URL = re.compile(r"https?://|www\.", re.IGNORECASE)
|
|
DATA_EMAIL = re.compile(r"\b[\w.+-]+@[\w.-]+\.[a-z]{2,}\b", re.IGNORECASE)
|
|
DATA_PHONE = re.compile(
|
|
r"(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b|\b\d{3}[-.\s]\d{3}[-.\s]\d{4}\b"
|
|
)
|
|
DATA_MONEY = re.compile(
|
|
r"\$\s*[\d,]+(?:\.\d{2})?|\b[\d,]+\s*(?:cad|usd|dollars?)\b", re.IGNORECASE
|
|
)
|
|
DATA_DATE = re.compile(
|
|
r"\b(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|"
|
|
r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{1,2}(?:,?\s+\d{4})?)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
DATA_ADDRESS = re.compile(
|
|
r"\b\d{1,5}\s+\w+(?:\s+\w+){0,3}\s+"
|
|
r"(?:st|street|ave|avenue|rd|road|dr|drive|blvd|way|crt|court|lane|ln|unit|apt|#)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
DATA_ACCOUNT = re.compile(
|
|
r"\b(?:account|policy|invoice|receipt|confirmation|booking|reference|"
|
|
r"transit|etransfer|sin|passport)\s*#?\s*[:#]?\s*[\w-]{4,}\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
CHAT_PATTERNS = re.compile(
|
|
r"\b(?:should we|what do you think|let me know|any thoughts|sounds good|"
|
|
r"i think we|we could|maybe we|let's\b|can you\b|do you want|"
|
|
r"worth it|decided to|going to\b|planning to|wondering if)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
NAME_LIST_TOPIC = re.compile(
|
|
r"\b(?:baby\s+names?|girl\s+names?|boy\s+names?|name\s+list|names?\s+we\s+like|"
|
|
r"favorite\s+names?|shortlist|hebrew\s+name|yiddish\s+name|middle\s+name)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
INFORMATIVE_FILE = re.compile(
|
|
r"invoice|receipt|lease|statement|bill|contract|insurance|tax|form|report|"
|
|
r"prescription|diagnosis|lab|result|scan|document|agreement|deed|mortgage|"
|
|
r"hoa|permit|inspection|warranty|quote|estimate|spreadsheet|\.pdf|\.doc|\.xls",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
CASUAL_IMAGE = re.compile(
|
|
r"^image\.|screenshot|slack-img|photo|img_\d|snap|slack-image|mvimg",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
PROPERTY_CHANNELS = {
|
|
"284richmond",
|
|
"45ritson",
|
|
"122harmony",
|
|
"6keefer",
|
|
"153niagara",
|
|
"banking",
|
|
"realestate",
|
|
"airbnb",
|
|
"york",
|
|
"viewings",
|
|
}
|
|
|
|
KIDS_CHANNELS = {
|
|
"zoey",
|
|
"izik",
|
|
"parenting",
|
|
"moishe",
|
|
"baby",
|
|
"after-school-activities",
|
|
}
|
|
|
|
FAMILY_CHANNELS = {"levkin", "medical", "purchase"}
|
|
|
|
AUTO_CHANNELS = PROPERTY_CHANNELS | KIDS_CHANNELS | FAMILY_CHANNELS
|
|
|
|
IMAGE_EXTS = {"png", "jpg", "jpeg", "gif", "webp", "bmp", "svg", "heic"}
|
|
|
|
|
|
def draft_group_key(name: str) -> str:
|
|
n = name.lower().strip()
|
|
n = re.sub(r"^draft[\s_\-]+", "", n)
|
|
n = re.sub(r"[\s_\-]*(final|v\d+|rev\d*|\(\d+\)|copy)", "", n)
|
|
n = re.sub(r"\.[a-z0-9]{1,5}$", "", n)
|
|
n = re.sub(r"\s+", " ", n).strip()
|
|
return n or name.lower()
|
|
|
|
|
|
def is_image(file: dict) -> bool:
|
|
mt = (file.get("mimetype") or "").lower()
|
|
if mt.startswith("image/"):
|
|
return True
|
|
name = (file.get("name") or "").lower()
|
|
ext = name.rsplit(".", 1)[-1] if "." in name else ""
|
|
return ext in IMAGE_EXTS
|
|
|
|
|
|
def is_doc(file: dict) -> bool:
|
|
name = (file.get("name") or "").lower()
|
|
ext = name.rsplit(".", 1)[-1] if "." in name else ""
|
|
return ext in {"pdf", "doc", "docx", "xls", "xlsx", "csv", "ppt", "pptx"}
|
|
|
|
|
|
def is_draft_file(file: dict) -> bool:
|
|
return "draft" in (file.get("name") or "").lower()
|
|
|
|
|
|
_DATA_PATTERNS = (
|
|
DATA_URL,
|
|
DATA_EMAIL,
|
|
DATA_PHONE,
|
|
DATA_MONEY,
|
|
DATA_DATE,
|
|
DATA_ADDRESS,
|
|
DATA_ACCOUNT,
|
|
)
|
|
|
|
|
|
def data_signal_count(text: str) -> int:
|
|
return sum(1 for p in _DATA_PATTERNS if p.search(text))
|
|
|
|
|
|
def has_data_signals(text: str) -> bool:
|
|
return data_signal_count(text) > 0
|
|
|
|
|
|
def text_is_factual(text: str) -> bool:
|
|
"""Require hard facts, not a lone link in chat."""
|
|
if not text or not text.strip():
|
|
return False
|
|
if looks_conversational(text):
|
|
return False
|
|
n = data_signal_count(text)
|
|
if n >= 2:
|
|
return True
|
|
if DATA_PHONE.search(text) or DATA_MONEY.search(text) or DATA_EMAIL.search(text):
|
|
return True
|
|
if DATA_ADDRESS.search(text) and DATA_DATE.search(text):
|
|
return True
|
|
return False
|
|
|
|
|
|
def looks_conversational(text: str) -> bool:
|
|
if not text or NOISE_TEXT.match(text.strip()):
|
|
return True
|
|
if CHAT_PATTERNS.search(text):
|
|
return True
|
|
if text.count("?") >= 2 and len(text) < 500:
|
|
return True
|
|
# Short back-and-forth without hard facts.
|
|
if len(text) < 80 and not has_data_signals(text):
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_name_list_message(text: str) -> bool:
|
|
if not text or not text.strip():
|
|
return False
|
|
if NAME_LIST_TOPIC.search(text):
|
|
return True
|
|
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
|
|
if len(lines) >= 4:
|
|
name_lines = sum(
|
|
1
|
|
for ln in lines
|
|
if re.match(r"^[\s•\-\d.]*[A-Z][a-z]{2,}(?:\s+[A-Z][a-z]{2,})?\s*$", ln)
|
|
)
|
|
if name_lines >= 3:
|
|
return True
|
|
return False
|
|
|
|
|
|
def informative_filename(name: str) -> bool:
|
|
return bool(INFORMATIVE_FILE.search(name or ""))
|
|
|
|
|
|
def informative_image(file: dict, text: str) -> bool:
|
|
name = file.get("name") or ""
|
|
if CASUAL_IMAGE.search(name):
|
|
return text_is_factual(text)
|
|
if informative_filename(name):
|
|
return True
|
|
return text_is_factual(text)
|
|
|
|
|
|
def build_latest_draft_winners(export, channel: str) -> tuple[set[str], dict[str, set[str]]]:
|
|
groups: dict[str, list[tuple[float, str, str]]] = defaultdict(list)
|
|
for msg in export.iter_channel_messages(channel):
|
|
for f in msg.get("files") or []:
|
|
name = f.get("name") or ""
|
|
if not is_draft_file(f):
|
|
continue
|
|
try:
|
|
ts = float(str(msg.get("ts", "0")).split(".")[0])
|
|
except (TypeError, ValueError):
|
|
ts = 0.0
|
|
fid = f.get("id") or name
|
|
groups[draft_group_key(name)].append((ts, msg["id"], fid))
|
|
|
|
msg_ids: set[str] = set()
|
|
files_by_msg: dict[str, set[str]] = defaultdict(set)
|
|
for items in groups.values():
|
|
_ts, msg_id, fid = max(items, key=lambda x: x[0])
|
|
msg_ids.add(msg_id)
|
|
files_by_msg[msg_id].add(fid)
|
|
return msg_ids, files_by_msg
|
|
|
|
|
|
def should_import(
|
|
msg: dict,
|
|
channel: str,
|
|
*,
|
|
latest_draft_msgs: set[str],
|
|
) -> bool:
|
|
subtype = msg.get("subtype")
|
|
text = (msg.get("text") or "").strip()
|
|
files = msg.get("files") or []
|
|
msg_id = msg["id"]
|
|
|
|
if subtype in SKIP_SUBTYPES and not files:
|
|
return False
|
|
if msg.get("is_bot") and not files:
|
|
return False
|
|
|
|
# --- Attachments (primary signal) ---
|
|
if files:
|
|
if msg_id in latest_draft_msgs:
|
|
return True
|
|
for f in files:
|
|
if is_draft_file(f):
|
|
continue
|
|
if is_doc(f):
|
|
return True
|
|
if informative_filename(f.get("name") or ""):
|
|
return True
|
|
if is_image(f) and informative_image(f, text):
|
|
return True
|
|
return False
|
|
|
|
# --- Text without files: facts only, not chat ---
|
|
if channel in KIDS_CHANNELS and is_name_list_message(text):
|
|
return True
|
|
|
|
return text_is_factual(text)
|
|
|
|
|
|
def msg_selection(
|
|
msg: dict,
|
|
*,
|
|
latest_draft_files: set[str] | None = None,
|
|
) -> dict:
|
|
text = (msg.get("text") or "").strip()
|
|
files: dict[str, bool] = {}
|
|
for f in msg.get("files") or []:
|
|
fid = f.get("id") or f.get("name")
|
|
if not fid:
|
|
continue
|
|
if is_draft_file(f):
|
|
if latest_draft_files and fid in latest_draft_files:
|
|
files[fid] = True
|
|
continue
|
|
if is_doc(f) or informative_filename(f.get("name") or ""):
|
|
files[fid] = True
|
|
elif is_image(f) and informative_image(f, text):
|
|
files[fid] = True
|
|
return {
|
|
"action": "import",
|
|
"checked": True,
|
|
"merge_target": "",
|
|
"notes": "auto-cherry-pick",
|
|
"files": files,
|
|
}
|
|
|
|
|
|
def channel_entry_skip(notes: str) -> dict:
|
|
return {
|
|
"import": False,
|
|
"status": "skip",
|
|
"reviewed": True,
|
|
"import_mode": "all",
|
|
"notes": notes,
|
|
"mattermost_channel": "",
|
|
"mattermost_team": "",
|
|
"messages": {},
|
|
}
|
|
|
|
|
|
def channel_shell() -> dict:
|
|
return {
|
|
"import": True,
|
|
"status": "partial",
|
|
"reviewed": True,
|
|
"import_mode": "pick",
|
|
"notes": "auto-cherry-pick — data/files only",
|
|
"mattermost_channel": "",
|
|
"mattermost_team": "",
|
|
"messages": {},
|
|
}
|
|
|
|
|
|
def reset_all_picks(manifest: dict) -> None:
|
|
channels = manifest.setdefault("channels", {})
|
|
for name, entry in channels.items():
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
entry["messages"] = {}
|
|
if name in RETURN_CHANNELS:
|
|
channels[name] = channel_entry_skip(
|
|
"Returned — not importing (29wilson / 520steeles)"
|
|
)
|
|
continue
|
|
if entry.get("status") == "skip":
|
|
continue
|
|
entry["import"] = False
|
|
entry["status"] = "undecided"
|
|
entry["reviewed"] = False
|
|
entry["import_mode"] = "all"
|
|
|
|
|
|
def main() -> int:
|
|
dry_run = "--dry-run" in sys.argv
|
|
reset_only = "--reset" in sys.argv
|
|
export_path = DEFAULT_EXPORT
|
|
for arg in sys.argv[1:]:
|
|
if not arg.startswith("-") and Path(arg).expanduser().is_dir():
|
|
export_path = Path(arg).expanduser().resolve()
|
|
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
from slack_export import SlackExport
|
|
|
|
export = SlackExport(export_path)
|
|
manifest: dict = {"channels": {}, "users": {}}
|
|
if MANIFEST_PATH.is_file():
|
|
manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
|
|
|
|
reset_all_picks(manifest)
|
|
print("Cleared picks — data/files only mode.")
|
|
for name in sorted(RETURN_CHANNELS):
|
|
print(f" #{name} → skip")
|
|
|
|
if reset_only:
|
|
if not dry_run:
|
|
manifest["updated_at"] = datetime.now(tz=timezone.utc).isoformat()
|
|
MANIFEST_PATH.write_text(
|
|
json.dumps(manifest, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return 0
|
|
|
|
channels = manifest.setdefault("channels", {})
|
|
stats: dict[str, int] = {}
|
|
|
|
for ch_name in sorted(AUTO_CHANNELS):
|
|
if ch_name in RETURN_CHANNELS:
|
|
continue
|
|
folder = export.channel_folder(ch_name)
|
|
if not folder:
|
|
continue
|
|
entry = channels.setdefault(ch_name, channel_shell())
|
|
entry["messages"] = {}
|
|
if entry.get("status") == "import" and entry.get("import_mode", "all") != "pick":
|
|
print(f"#{ch_name}: whole-channel import — left unchanged")
|
|
continue
|
|
entry["import"] = True
|
|
entry["status"] = "partial"
|
|
entry["reviewed"] = True
|
|
entry["import_mode"] = "pick"
|
|
|
|
latest_draft_msgs, draft_files_by_msg = build_latest_draft_winners(export, ch_name)
|
|
messages = entry.setdefault("messages", {})
|
|
picked = 0
|
|
for msg in export.iter_channel_messages(ch_name):
|
|
if not should_import(msg, ch_name, latest_draft_msgs=latest_draft_msgs):
|
|
continue
|
|
messages[msg["id"]] = msg_selection(
|
|
msg, latest_draft_files=draft_files_by_msg.get(msg["id"])
|
|
)
|
|
picked += 1
|
|
stats[ch_name] = picked
|
|
print(f"#{ch_name}: {picked} msgs")
|
|
|
|
total = sum(stats.values())
|
|
print(f"\nTotal: {total} messages (data/files only)")
|
|
|
|
if dry_run:
|
|
print("(dry-run — manifest not written)")
|
|
return 0
|
|
|
|
manifest["updated_at"] = datetime.now(tz=timezone.utc).isoformat()
|
|
note = manifest.get("global_notes") or ""
|
|
stamp = "auto-cherry-pick v3 — data/files only, no conversation"
|
|
if stamp not in note:
|
|
manifest["global_notes"] = (note + "\n" + stamp).strip()
|
|
MANIFEST_PATH.write_text(
|
|
json.dumps(manifest, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(f"Wrote {MANIFEST_PATH}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|