Spaces:
Sleeping
Sleeping
| """Folder Validator. | |
| Scans a user-supplied dossier folder, classifies each readable file into a | |
| source type, and reports a dossier strength score that reflects the | |
| overall quality and completeness of the evidence collection. | |
| The validator is intentionally flexible: it does not require any single | |
| file (resume, profile, etc.) to be present. A dossier is valid as long | |
| as at least one readable supported file exists. | |
| Filename stems and a small text sample from plain-text files are used | |
| for classification signals. Full PDF / DOCX / image content is not read | |
| here — see app/services/dossier_reader.py for full extraction. | |
| """ | |
| from __future__ import annotations | |
| import codecs | |
| import json | |
| import os | |
| import re | |
| from dataclasses import dataclass, field | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Optional | |
| from app.models.schemas import ( | |
| SOURCE_PRIORITY, | |
| SOURCE_TYPE_LABELS, | |
| SourceType, | |
| ) | |
| from app.utils.file_utils import iter_contained_files | |
| SUPPORTED_EXTENSIONS: frozenset[str] = frozenset( | |
| {".pdf", ".docx", ".txt", ".md", ".json", ".csv", ".png", ".jpg", ".jpeg"} | |
| ) | |
| TEXT_LIKE_EXTENSIONS: frozenset[str] = frozenset({".txt", ".md", ".json", ".csv"}) | |
| IMAGE_EXTENSIONS: frozenset[str] = frozenset({".png", ".jpg", ".jpeg"}) | |
| WARNING_SCORE_THRESHOLD = 30 | |
| THIN_DOSSIER_THRESHOLD = 40 | |
| STRENGTH_LABELS: tuple[tuple[int, str], ...] = ( | |
| (80, "Strong evidence collection"), | |
| (60, "Good evidence collection"), | |
| (40, "Usable but needs stronger proof"), | |
| (0, "Thin dossier"), | |
| ) | |
| # Bucket key, label, weight. Scoring is rule-driven (see _score below) so | |
| # different buckets can use different signals (source type vs. keyword vs. | |
| # diversity count). | |
| _BUCKETS: tuple[tuple[str, str, int], ...] = ( | |
| ("structured_profile_data", "Structured profile data", 20), | |
| ("positioning_and_offer", "Positioning and offer clarity", 15), | |
| ("work_history_background", "Work history and background", 15), | |
| ("skills_tools_services", "Skills, tools, and service evidence", 15), | |
| ("proof_credibility", "Proof and credibility", 15), | |
| ("proposal_preferences_pricing", "Proposal preferences and pricing", 10), | |
| ("completeness_diversity", "Completeness and diversity of sources", 10), | |
| ) | |
| # Filename / content keyword patterns mapped to source types. The order | |
| # matters: more specific patterns are checked before general ones. | |
| _FILENAME_RULES: tuple[tuple[SourceType, tuple[str, ...]], ...] = ( | |
| ("linkedin_optimization", ("linkedin optimization", "linkedin optimisation", "linkedin rewrite")), | |
| ("skillarbitrage_dossier_roadmap", ("skillarbitrage", "dossier", "roadmap")), | |
| ("offer_blueprint", ("offer blueprint", "offer-blueprint", "offer_blueprint", "blueprint", "selected offer", "offer bank")), | |
| # Only the Upwork Profile Optimisation document itself: it outranks the | |
| # Dossier and stands in for the chosen offer. A file merely named for | |
| # Upwork (past proposals, reviews) is placed by the rules below. | |
| ("upwork_profile", ("upwork profile optimi", "upwork optimi")), | |
| ("linkedin_profile", ("linkedin",)), | |
| ("discovery_call_transcript", ("transcript", "discovery call", "discovery-call", "discovery_call", "call notes")), | |
| ("resume_or_cv", ("resume", "curriculum vitae", "cv")), | |
| ("testimonial_or_review", ("testimonial", "review", "feedback", "endorsement", "recommendation letter")), | |
| ("portfolio_or_case_study", ("portfolio", "case study", "case-study", "case_study", "work sample", "work_sample", "sample")), | |
| ("past_proposal", ("proposal sample", "proposal-sample", "past proposal", "previous proposal", "proposal example", "proposal")), | |
| ("pricing_or_service_package", ("pricing", "rate card", "rate-card", "service package", "service-package", "service stack", "service-stack", "package", "rates")), | |
| ("certification_or_course", ("certificate", "certification", "course completion", "course-completion", "course_completion", "course", "credential")), | |
| ("client_research", ("client avatar", "client-avatar", "target client", "target-client", "icp ", " icp", "icp_", "ideal client")), | |
| ("niche_research", ("niche", "market research", "market-research", "industry research", "industry-research")), | |
| ("personal_branding", ("personal brand", "personal-brand", "headline", "bio", "about section", "about-section", "branding")), | |
| ("strategy_document", ("strategy", "positioning", "plan")), | |
| ("notes_or_misc_profile_context", ("notes", "scratch")), | |
| # Named for Upwork and nothing more ("Upwork.pdf", an exported profile): | |
| # a general profile document, low priority. The reader still files a | |
| # PDF or Word file as the optimisation document when its cover says so. | |
| ("generic_profile_document", ("upwork",)), | |
| ) | |
| class FileRecord: | |
| """One file's place in the dossier.""" | |
| relative_path: str | |
| extension: str | |
| supported: bool | |
| readable: bool | |
| source_type: Optional[SourceType] = None | |
| source_priority: Optional[int] = None | |
| extraction_status: str = "pending" | |
| modified_at: Optional[str] = None | |
| note: Optional[str] = None | |
| class FolderValidationResult: | |
| folder: str | |
| exists: bool = False | |
| files: list[FileRecord] = field(default_factory=list) | |
| last_modified: dict[str, str] = field(default_factory=dict) | |
| strength_score: int = 0 | |
| score_breakdown: dict[str, int] = field(default_factory=dict) | |
| source_type_counts: dict[str, int] = field(default_factory=dict) | |
| missing_categories: list[str] = field(default_factory=list) | |
| issues: list[str] = field(default_factory=list) | |
| warnings: list[str] = field(default_factory=list) | |
| notes: list[str] = field(default_factory=list) | |
| # --- backwards-compatible / convenience views --- | |
| def readable_files(self) -> list[str]: | |
| return [ | |
| f.relative_path for f in self.files if f.supported and f.readable | |
| ] | |
| def unsupported_files(self) -> list[str]: | |
| return [f.relative_path for f in self.files if not f.supported] | |
| def readable_count(self) -> int: | |
| return len(self.readable_files) | |
| def total_files(self) -> int: | |
| return len(self.files) | |
| def supported_count(self) -> int: | |
| return sum(1 for f in self.files if f.supported) | |
| def unsupported_count(self) -> int: | |
| return sum(1 for f in self.files if not f.supported) | |
| def is_empty(self) -> bool: | |
| return not self.files | |
| def below_threshold(self) -> bool: | |
| return self.strength_score < WARNING_SCORE_THRESHOLD | |
| def can_continue(self) -> bool: | |
| return self.exists and self.readable_count > 0 | |
| def strength_label(self) -> str: | |
| for threshold, label in STRENGTH_LABELS: | |
| if self.strength_score >= threshold: | |
| return label | |
| return "Thin dossier" | |
| # --------------------------------------------------------------------------- | |
| # Source classification | |
| # --------------------------------------------------------------------------- | |
| _STRUCTURED_PROFILE_KEYS: frozenset[str] = frozenset( | |
| { | |
| "name", | |
| "title", | |
| "positioning", | |
| "skills", | |
| "tools", | |
| "services", | |
| "service_stack", | |
| "target_client", | |
| "work_history", | |
| "proposal_preferences", | |
| "selected_offer", | |
| "offer", | |
| "pricing", | |
| "deliverables", | |
| "industries", | |
| } | |
| ) | |
| def _is_placeholder(value: object) -> bool: | |
| """Return True if a JSON value looks like an unfilled template slot.""" | |
| if value is None: | |
| return True | |
| if isinstance(value, str): | |
| s = value.strip().lower() | |
| if not s: | |
| return True | |
| return s in { | |
| "tbd", | |
| "todo", | |
| "todo:", | |
| "placeholder", | |
| "fill in", | |
| "fill-in", | |
| "fillme", | |
| "n/a", | |
| "na", | |
| "example", | |
| "your name", | |
| "your title", | |
| "<insert>", | |
| "<replace>", | |
| } | |
| if isinstance(value, (list, dict)): | |
| return len(value) == 0 | |
| return False | |
| def _classify_json(data: object) -> tuple[SourceType, Optional[str]]: | |
| """Decide which JSON source type a parsed payload looks like.""" | |
| if not isinstance(data, dict): | |
| return "generic_profile_document", None | |
| # Map each lowercased profile key back to the value under the original | |
| # (possibly differently-cased) JSON key, so "Name"/"Skills" are matched. | |
| lower_to_value: dict[str, object] = {} | |
| for raw_key, value in data.items(): | |
| if isinstance(raw_key, str): | |
| lower_to_value.setdefault(raw_key.lower(), value) | |
| profile_hits = set(lower_to_value) & _STRUCTURED_PROFILE_KEYS | |
| if profile_hits: | |
| # Treat as a real profile only if at least one of the PROFILE-SHAPED | |
| # keys actually has data. A template whose name/skills/etc. are all | |
| # empty is a blank template even if some unrelated metadata key | |
| # (e.g. "version"/"schema") is populated — so we check profile_hits, | |
| # not every key in the document. | |
| profile_has_data = any( | |
| not _is_placeholder(lower_to_value.get(k)) for k in profile_hits | |
| ) | |
| if profile_has_data: | |
| return "structured_profile_json", None | |
| return "dossier_template_json", "All structured fields look empty." | |
| return "generic_profile_document", None | |
| # Byte-order marks, longest first: the UTF-32 LE mark begins with the | |
| # UTF-16 LE one. Each codec named here drops the mark as it decodes. | |
| _BOM_ENCODINGS: tuple[tuple[bytes, str], ...] = ( | |
| (codecs.BOM_UTF32_LE, "utf-32"), | |
| (codecs.BOM_UTF32_BE, "utf-32"), | |
| (codecs.BOM_UTF8, "utf-8-sig"), | |
| (codecs.BOM_UTF16_LE, "utf-16"), | |
| (codecs.BOM_UTF16_BE, "utf-16"), | |
| ) | |
| # One whole UTF-8 multi-byte character. | |
| _UTF8_CHARACTER = re.compile( | |
| rb"[\xc2-\xdf][\x80-\xbf]|[\xe0-\xef][\x80-\xbf]{2}|[\xf0-\xf4][\x80-\xbf]{3}" | |
| ) | |
| def _text_encoding(raw: bytes) -> str: | |
| """The encoding a text file's byte-order mark (or NUL bytes) shows, else UTF-8.""" | |
| for bom, encoding in _BOM_ENCODINGS: | |
| if raw.startswith(bom): | |
| return encoding | |
| # UTF-16 saved without a mark: in Latin text every other byte is NUL. | |
| head = raw[:4096] | |
| half = len(head) // 2 | |
| even, odd = head[0::2].count(0), head[1::2].count(0) | |
| if half >= 2 and odd * 2 >= half and even * 10 < half: | |
| return "utf-16-le" | |
| if half >= 2 and even * 2 >= half and odd * 10 < half: | |
| return "utf-16-be" | |
| return "utf-8" | |
| def decode_text(raw: bytes) -> tuple[str, bool]: | |
| """Decode a text file's bytes; returns ``(text, lossy)``. | |
| Windows tools save text with a byte-order mark (UTF-8 in older Notepad, | |
| UTF-16 as "Unicode" or Excel's "Unicode Text"), or as Windows-1252 | |
| ("ANSI"). Read as plain UTF-8, the mark made JSON unparseable and UTF-16 | |
| came out as letters split by NUL bytes. ``lossy`` is True when bytes | |
| that fit no encoding were dropped. | |
| """ | |
| encoding = _text_encoding(raw) | |
| try: | |
| return raw.decode(encoding), False | |
| except UnicodeDecodeError: | |
| pass | |
| if encoding == "utf-8" and not _UTF8_CHARACTER.search(raw): | |
| # Not one UTF-8 character in it: Windows-1252 ("ANSI") text. | |
| try: | |
| return raw.decode("cp1252"), False | |
| except UnicodeDecodeError: | |
| pass | |
| return raw.decode(encoding, errors="ignore"), True | |
| def _read_text_sample(path: Path, limit_bytes: int = 8192) -> str: | |
| try: | |
| with open(path, "rb") as fh: | |
| chunk = fh.read(limit_bytes) | |
| except OSError: | |
| return "" | |
| return decode_text(chunk)[0] | |
| def _normalize_for_match(text: str) -> str: | |
| return text.lower().replace("_", " ").replace("-", " ") | |
| def classify_source( | |
| path: Path, | |
| text_sample: str = "", | |
| json_payload: object = None, | |
| ) -> tuple[SourceType, Optional[str]]: | |
| """Classify a single file as the best-matching source type. | |
| Returns (source_type, optional_note). The note is surfaced to the UI | |
| when classification adds useful context (e.g., template JSON). | |
| Unknown but readable files fall through to ``unknown_supported_file``. | |
| """ | |
| ext = path.suffix.lower() | |
| # JSON gets a content-aware classifier. | |
| if ext == ".json" and json_payload is not None: | |
| return _classify_json(json_payload) | |
| if ext == ".json": | |
| return "generic_profile_document", None | |
| stem = _normalize_for_match(path.stem) | |
| sample = _normalize_for_match(text_sample[:2048]) if text_sample else "" | |
| for source_type, patterns in _FILENAME_RULES: | |
| for pattern in patterns: | |
| if pattern in stem: | |
| return source_type, None | |
| # Content-only fallback: look in the text sample for a few high-signal hits. | |
| for source_type, patterns in _FILENAME_RULES: | |
| for pattern in patterns: | |
| if pattern in sample: | |
| return source_type, None | |
| # Image files with no naming signal are still useful as work samples. | |
| if ext in IMAGE_EXTENSIONS: | |
| return "portfolio_or_case_study", "Image file — treated as a sample." | |
| return "unknown_supported_file", None | |
| # --------------------------------------------------------------------------- | |
| # Strength scoring | |
| # --------------------------------------------------------------------------- | |
| _SOURCE_TO_BUCKETS: dict[str, tuple[str, ...]] = { | |
| "structured_profile_json": ("structured_profile_data",), | |
| "dossier_template_json": (), | |
| "skillarbitrage_dossier_roadmap": ("structured_profile_data", "positioning_and_offer"), | |
| "linkedin_optimization": ("positioning_and_offer", "work_history_background"), | |
| "offer_blueprint": ("positioning_and_offer",), | |
| "upwork_profile": ("work_history_background", "skills_tools_services"), | |
| "linkedin_profile": ("work_history_background",), | |
| "discovery_call_transcript": ("work_history_background", "positioning_and_offer"), | |
| "resume_or_cv": ("work_history_background",), | |
| "pricing_or_service_package": ("proposal_preferences_pricing", "skills_tools_services"), | |
| "portfolio_or_case_study": ("proof_credibility", "skills_tools_services"), | |
| "testimonial_or_review": ("proof_credibility",), | |
| "certification_or_course": ("proof_credibility", "skills_tools_services"), | |
| "past_proposal": ("proposal_preferences_pricing",), | |
| "client_research": ("positioning_and_offer",), | |
| "niche_research": ("positioning_and_offer",), | |
| "personal_branding": ("positioning_and_offer",), | |
| "strategy_document": ("positioning_and_offer",), | |
| "notes_or_misc_profile_context": (), | |
| "generic_profile_document": (), | |
| "unknown_supported_file": (), | |
| } | |
| _SKILL_TOOL_KEYWORDS = ( | |
| "skill", "skills", "tool", "tools", "stack", "tech stack", | |
| "service", "services", "deliverable", "deliverables", "capability", | |
| "capabilities", "expertise", "specialty", "speciality", | |
| ) | |
| _PROOF_KEYWORDS = ( | |
| "result", "results", "metric", "kpi", "roi", "growth", "revenue", | |
| "uplift", "conversion", "increase", "decrease", "%", "$", | |
| "testimonial", "review", "feedback", | |
| ) | |
| _PRICING_KEYWORDS = ( | |
| "rate", "rates", "pricing", "package", "tone", "voice", | |
| "deposit", "retainer", "hourly", "fixed", "budget", "fee", | |
| ) | |
| def _bucket_count(records: list[FileRecord], bucket_key: str) -> int: | |
| return sum( | |
| 1 | |
| for r in records | |
| if r.source_type and bucket_key in _SOURCE_TO_BUCKETS.get(r.source_type, ()) | |
| ) | |
| def _score( | |
| records: list[FileRecord], | |
| text_corpus: str, | |
| ) -> tuple[int, dict[str, int]]: | |
| """Compute the 7-bucket strength score (max 100).""" | |
| breakdown: dict[str, int] = {} | |
| total = 0 | |
| has_structured = any( | |
| r.source_type == "structured_profile_json" for r in records | |
| ) | |
| breakdown["structured_profile_data"] = 20 if has_structured else 0 | |
| has_positioning = _bucket_count(records, "positioning_and_offer") > 0 | |
| breakdown["positioning_and_offer"] = 15 if has_positioning else 0 | |
| has_history = _bucket_count(records, "work_history_background") > 0 | |
| breakdown["work_history_background"] = 15 if has_history else 0 | |
| has_skills_doc = _bucket_count(records, "skills_tools_services") > 0 | |
| skills_in_corpus = any(kw in text_corpus for kw in _SKILL_TOOL_KEYWORDS) | |
| breakdown["skills_tools_services"] = ( | |
| 15 if (has_skills_doc or skills_in_corpus) else 0 | |
| ) | |
| has_proof_doc = _bucket_count(records, "proof_credibility") > 0 | |
| proof_in_corpus = any(kw in text_corpus for kw in _PROOF_KEYWORDS) | |
| breakdown["proof_credibility"] = ( | |
| 15 if (has_proof_doc or proof_in_corpus) else 0 | |
| ) | |
| has_pricing_doc = _bucket_count(records, "proposal_preferences_pricing") > 0 | |
| pricing_in_corpus = any(kw in text_corpus for kw in _PRICING_KEYWORDS) | |
| breakdown["proposal_preferences_pricing"] = ( | |
| 10 if (has_pricing_doc or pricing_in_corpus) else 0 | |
| ) | |
| useful_source_types = { | |
| r.source_type | |
| for r in records | |
| if r.source_type | |
| and r.source_type | |
| not in { | |
| "unknown_supported_file", | |
| "dossier_template_json", | |
| } | |
| } | |
| breakdown["completeness_diversity"] = ( | |
| 10 if len(useful_source_types) >= 3 else 0 | |
| ) | |
| total = sum(breakdown.values()) | |
| return min(total, 100), breakdown | |
| def score_rubric_labels() -> dict[str, str]: | |
| return {key: label for key, label, _w in _BUCKETS} | |
| # --------------------------------------------------------------------------- | |
| # Strength from what was read | |
| # --------------------------------------------------------------------------- | |
| # | |
| # ``validate`` scores a folder before anything in it is read: it sees file | |
| # names, and the contents of plain-text files only, so a 36-page dossier PDF | |
| # named "Dossier.pdf" scored 30/100 and was called thin. Once the files have | |
| # been read into proof points, ``score_evidence`` scores what they actually | |
| # hold, on the same seven buckets and weights. | |
| # The kinds of claim that fill each bucket. | |
| _EVIDENCE_BUCKETS: dict[str, frozenset[str]] = { | |
| "positioning_and_offer": frozenset({"positioning", "selected_offer", "target_client"}), | |
| "work_history_background": frozenset({"experience", "work_history", "education", "achievement"}), | |
| "skills_tools_services": frozenset({"skill", "tool", "service", "deliverable"}), | |
| "proof_credibility": frozenset({"metric", "testimonial", "project", "portfolio", "certification"}), | |
| "proposal_preferences_pricing": frozenset({"pricing", "proposal_preference", "guarantee"}), | |
| } | |
| # The profile facts a proposal can draw on; how many are covered fills the | |
| # "structured profile" bucket. | |
| _PROFILE_FACT_GROUPS: tuple[frozenset[str], ...] = ( | |
| frozenset({"identity"}), | |
| frozenset({"positioning", "selected_offer"}), | |
| frozenset({"target_client"}), | |
| frozenset({"service", "deliverable"}), | |
| frozenset({"skill", "tool"}), | |
| frozenset({"experience", "work_history", "achievement"}), | |
| frozenset({"education", "certification"}), | |
| frozenset({"language"}), | |
| frozenset({"pricing"}), | |
| ) | |
| _PROFILE_FACTS_FOR_FULL_MARKS = 7 | |
| _DISTINCT_CLAIMS_FOR_DIVERSITY = 25 | |
| # What each empty bucket means, in words a student understands. | |
| MISSING_IN_PLAIN_WORDS: dict[str, str] = { | |
| "structured_profile_data": "basic profile details (education, languages, tools)", | |
| "positioning_and_offer": "who you help and what you offer", | |
| "work_history_background": "your work history or background", | |
| "skills_tools_services": "your skills, tools and services", | |
| "proof_credibility": "past results, samples or testimonials", | |
| "proposal_preferences_pricing": "your rates or pricing", | |
| "completeness_diversity": "enough detail overall", | |
| } | |
| def score_evidence(result: FolderValidationResult, proofs: list) -> FolderValidationResult: | |
| """Re-score ``result`` from the proof points read out of its files. | |
| The higher of the two scores stands, so a folder of well-named plain-text | |
| files is never marked down. ``missing_categories`` is rebuilt to name | |
| only what the files really lack. | |
| """ | |
| claim_types = {str(getattr(p, "claim_type", "") or "") for p in proofs or []} | |
| breakdown: dict[str, int] = {} | |
| facts = sum(1 for group in _PROFILE_FACT_GROUPS if group & claim_types) | |
| breakdown["structured_profile_data"] = round( | |
| 20 * min(facts, _PROFILE_FACTS_FOR_FULL_MARKS) / _PROFILE_FACTS_FOR_FULL_MARKS | |
| ) | |
| weights = {key: weight for key, _label, weight in _BUCKETS} | |
| for key, kinds in _EVIDENCE_BUCKETS.items(): | |
| breakdown[key] = weights[key] if kinds & claim_types else 0 | |
| distinct = {str(getattr(p, "claim_text", "") or "") for p in proofs or []} | |
| sources = {str(getattr(p, "source_type", "") or "") for p in proofs or []} | |
| breakdown["completeness_diversity"] = ( | |
| 10 if len(distinct) >= _DISTINCT_CLAIMS_FOR_DIVERSITY or len(sources) >= 3 else 0 | |
| ) | |
| total = min(sum(breakdown.values()), 100) | |
| if total >= result.strength_score: | |
| result.strength_score = total | |
| result.score_breakdown = breakdown | |
| result.missing_categories = _missing_categories(breakdown) | |
| # The pre-read warning about a low score no longer applies once the | |
| # content has been scored above it. | |
| if not result.below_threshold: | |
| result.warnings = [w for w in result.warnings if "strength score" not in w] | |
| return result | |
| def missing_in_plain_words(result: FolderValidationResult) -> list[str]: | |
| """The empty buckets of ``result``, phrased for a student.""" | |
| return [ | |
| MISSING_IN_PLAIN_WORDS[key] | |
| for key, awarded in (result.score_breakdown or {}).items() | |
| if not awarded and key in MISSING_IN_PLAIN_WORDS | |
| ] | |
| # --------------------------------------------------------------------------- | |
| # Main entry point | |
| # --------------------------------------------------------------------------- | |
| def _is_readable(path: Path) -> bool: | |
| return os.access(path, os.R_OK) | |
| # No profile nests anywhere near this deep. The JSON parser, and the code | |
| # that later walks a payload, recurse once per level, so a deeper file | |
| # raised RecursionError out of the Dossier screen instead of being unreadable. | |
| _JSON_MAX_DEPTH = 100 | |
| def _nested_deeper_than(payload: object, limit: int) -> bool: | |
| """True when lists and objects in ``payload`` nest more than ``limit`` deep. | |
| Walked with a stack, not recursion, for the reason above. | |
| """ | |
| stack: list[tuple[object, int]] = [(payload, 1)] | |
| while stack: | |
| value, depth = stack.pop() | |
| if isinstance(value, dict): | |
| children = value.values() | |
| elif isinstance(value, list): | |
| children = value | |
| else: | |
| continue | |
| if depth > limit: | |
| return True | |
| stack.extend((child, depth + 1) for child in children) | |
| return False | |
| def load_json_file(path: Path) -> object: | |
| """Parse a JSON file; the dossier reader uses this too. | |
| Raises ``OSError`` when the file cannot be read and ``ValueError`` when | |
| it is not usable JSON, including a file nested too deep to read safely. | |
| The text is decoded as :func:`decode_text` reads it. | |
| """ | |
| text, _lossy = decode_text(path.read_bytes()) | |
| try: | |
| payload = json.loads(text) | |
| except RecursionError: | |
| raise ValueError("JSON is nested too deeply") from None | |
| if _nested_deeper_than(payload, _JSON_MAX_DEPTH): | |
| raise ValueError("JSON is nested too deeply") | |
| return payload | |
| def _load_json_safely(path: Path) -> tuple[object, Optional[str]]: | |
| try: | |
| return load_json_file(path), None | |
| except (OSError, ValueError) as exc: | |
| return None, f"JSON unreadable: {exc.__class__.__name__}" | |
| def _build_corpus(records: list[FileRecord], samples: dict[str, str]) -> str: | |
| parts: list[str] = [] | |
| for r in records: | |
| parts.append(_normalize_for_match(Path(r.relative_path).stem)) | |
| sample = samples.get(r.relative_path) | |
| if sample: | |
| parts.append(_normalize_for_match(sample)) | |
| return "\n".join(parts) | |
| def _missing_categories(breakdown: dict[str, int]) -> list[str]: | |
| labels = score_rubric_labels() | |
| return [labels[key] for key, awarded in breakdown.items() if awarded == 0] | |
| def validate(folder_path: str | Path) -> FolderValidationResult: | |
| """Validate a dossier folder as a flexible evidence collection. | |
| Walks the folder, classifies each readable file into a source type, | |
| and computes a 0-100 strength score across seven evidence buckets. | |
| No assumption is made that any specific file type is mandatory — the | |
| only hard requirement is that at least one readable supported file | |
| exists. | |
| """ | |
| folder = Path(folder_path).expanduser() | |
| result = FolderValidationResult(folder=str(folder)) | |
| if not folder.exists(): | |
| result.issues.append(f"Folder does not exist: {folder}") | |
| return result | |
| if not folder.is_dir(): | |
| result.issues.append(f"Path is not a directory: {folder}") | |
| return result | |
| result.exists = True | |
| try: | |
| # Symlink-safe, containment-checked, count-bounded walk (shared with | |
| # the dossier reader). Symlinked entries and paths resolving outside | |
| # the folder are excluded so the validator never classifies or | |
| # samples content from outside the chosen folder. | |
| walk: dict = {} | |
| candidates = list(iter_contained_files(folder, stats=walk)) | |
| if walk.get("truncated"): | |
| result.warnings.append("Folder scan reached its safety limit; some files were not read. Choose a smaller dossier folder.") | |
| except OSError as exc: | |
| result.issues.append(f"Unable to walk folder: {exc}") | |
| return result | |
| samples: dict[str, str] = {} | |
| modified_times: list[float] = [] | |
| for path in candidates: | |
| rel = str(path.relative_to(folder)) | |
| ext = path.suffix.lower() | |
| record = FileRecord( | |
| relative_path=rel, | |
| extension=ext, | |
| supported=ext in SUPPORTED_EXTENSIONS, | |
| readable=_is_readable(path), | |
| ) | |
| try: | |
| mtime = path.stat().st_mtime | |
| record.modified_at = datetime.fromtimestamp(mtime).isoformat( | |
| timespec="seconds" | |
| ) | |
| modified_times.append(mtime) | |
| except OSError: | |
| pass | |
| if not record.supported: | |
| record.extraction_status = "unsupported" | |
| result.files.append(record) | |
| continue | |
| if not record.readable: | |
| record.extraction_status = "unreadable" | |
| result.issues.append(f"Unreadable file skipped: {rel}") | |
| result.files.append(record) | |
| continue | |
| text_sample = "" | |
| json_payload: object = None | |
| if ext in TEXT_LIKE_EXTENSIONS: | |
| text_sample = _read_text_sample(path) | |
| if ext == ".json": | |
| json_payload, json_warning = _load_json_safely(path) | |
| if json_warning: | |
| record.note = json_warning | |
| source_type, note = classify_source( | |
| path, text_sample=text_sample, json_payload=json_payload | |
| ) | |
| record.source_type = source_type | |
| record.source_priority = SOURCE_PRIORITY[source_type] | |
| record.extraction_status = "scanned" | |
| if note and not record.note: | |
| record.note = note | |
| if text_sample: | |
| samples[rel] = text_sample | |
| result.files.append(record) | |
| # Source-type counts. | |
| counts: dict[str, int] = {} | |
| for r in result.files: | |
| if r.source_type: | |
| counts[r.source_type] = counts.get(r.source_type, 0) + 1 | |
| result.source_type_counts = counts | |
| # Modified date range. | |
| if modified_times: | |
| earliest = datetime.fromtimestamp(min(modified_times)) | |
| latest = datetime.fromtimestamp(max(modified_times)) | |
| result.last_modified = { | |
| "earliest": earliest.isoformat(timespec="seconds"), | |
| "latest": latest.isoformat(timespec="seconds"), | |
| } | |
| readable_records = [r for r in result.files if r.supported and r.readable] | |
| corpus = _build_corpus(readable_records, samples) | |
| total, breakdown = _score(readable_records, corpus) | |
| result.strength_score = total | |
| result.score_breakdown = breakdown | |
| result.missing_categories = _missing_categories(breakdown) | |
| # --- Warnings --- | |
| if result.is_empty: | |
| result.warnings.append( | |
| "Folder is empty. You can continue, but the proposal will lack grounding." | |
| ) | |
| elif not readable_records: | |
| result.warnings.append( | |
| "No readable supported files were found. " | |
| f"Supported extensions: {', '.join(sorted(SUPPORTED_EXTENSIONS))}." | |
| ) | |
| if readable_records: | |
| useful_types = { | |
| r.source_type | |
| for r in readable_records | |
| if r.source_type | |
| and r.source_type | |
| not in {"unknown_supported_file", "dossier_template_json"} | |
| } | |
| only_resume_like = useful_types and useful_types.issubset( | |
| {"resume_or_cv", "linkedin_profile", "upwork_profile"} | |
| ) | |
| if only_resume_like: | |
| result.warnings.append( | |
| "Your dossier is readable but thin. Add offer details, proof " | |
| "files, portfolio samples, testimonials, or positioning " | |
| "documents for stronger proposals." | |
| ) | |
| if result.below_threshold and not result.is_empty: | |
| result.warnings.append( | |
| f"Dossier strength score is {result.strength_score}/100 " | |
| f"(below the recommended {WARNING_SCORE_THRESHOLD}). " | |
| "You can continue, but proposal grounding will be weak." | |
| ) | |
| if result.unsupported_count and readable_records: | |
| result.notes.append( | |
| f"{result.unsupported_count} file(s) skipped as unsupported." | |
| ) | |
| _ = SOURCE_TYPE_LABELS # exported for UI consumers | |
| return result | |