from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from transformers import AutoTokenizer, AutoModelForSeq2SeqLM from cachetools import TTLCache import hashlib import re import torch app = FastAPI(title="Local Summarizer") MODEL_NAME = "sshleifer/distilbart-cnn-12-6" MAX_INPUT_CHARS = 20000 tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_NAME) model.eval() device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) cache = TTLCache(maxsize=1024, ttl=60 * 60) class SummarizeRequest(BaseModel): text: str = Field(min_length=1, max_length=MAX_INPUT_CHARS) max_length: int = Field(default=160, ge=24, le=256) min_length: int = Field(default=45, ge=8, le=180) top_skills: int = Field(default=8, ge=3, le=12) def _key(text: str, max_length: int, min_length: int, top_skills: int) -> str: h = hashlib.sha256(text.encode("utf-8")).hexdigest() return f"{h}:{max_length}:{min_length}:{top_skills}" @app.get("/health") async def health(): return {"ok": True, "model": MODEL_NAME} _TECH = [ "python", "c#", "dotnet", ".net", "java", "javascript", "typescript", "react", "node", "sql", "postgres", "postgresql", "mysql", "sqlite", "mongodb", "redis", "aws", "azure", "gcp", "docker", "kubernetes", "terraform", "linux", "git", "ci/cd", "graphql", "rest", ] _SOFT = [ "communication", "collaboration", "teamwork", "problem solving", "leadership", "mentoring", "ownership", "initiative", "adaptability", "stakeholder management", "detail oriented", ] _TECH_PRIORITY = [ "python", "c#", ".net", "dotnet", "typescript", "javascript", "react", "node", "sql", "postgresql", "postgres", "mysql", "sqlite", "docker", "kubernetes", "aws", "azure", "gcp", "terraform", "graphql", "rest", "git", ] def _rank_tech_skills(skills): ordered = [] seen = set() for preferred in _TECH_PRIORITY: for skill in skills: if skill == preferred and skill not in seen: ordered.append(skill) seen.add(skill) for skill in skills: if skill not in seen: ordered.append(skill) seen.add(skill) return ordered def _strip_html(text: str) -> str: text = re.sub(r"<\s*br\s*/?>", "\n", text, flags=re.IGNORECASE) text = re.sub(r"", "\n", text, flags=re.IGNORECASE) text = re.sub(r"<[^>]+>", " ", text) return re.sub(r"\n{3,}", "\n\n", text).strip() def _extract_bullets(lines, max_items=8): out = [] for ln in lines: s = ln.strip() if not s: continue if re.match(r"^([-*]|\u2022)\s+", s): s = re.sub(r"^([-*]|\u2022)\s+", "", s).strip() if 3 <= len(s) <= 200: out.append(s) if len(out) >= max_items: break return out def _top_keywords(text: str, limit=6): words = re.findall(r"[a-zA-Z][a-zA-Z+#./-]{2,}", text.lower()) stop = { "with", "from", "that", "this", "will", "have", "your", "their", "about", "role", "team", "work", "experience", "skills", "requirements", "responsibilities", "company", "using", "ability", "years", } counts = {} for word in words: if word in stop or word in _TECH or word in _SOFT: continue counts[word] = counts.get(word, 0) + 1 ordered = sorted(counts.items(), key=lambda item: (-item[1], item[0])) return [word for word, _ in ordered[:limit]] def _role_focused_excerpt(text: str) -> dict: cleaned = _strip_html(text) lines = [ln.strip() for ln in cleaned.splitlines()] headings = { "responsibilities": ["responsibilities", "what you will do", "what you'll do", "the role", "your role", "you will"], "requirements": ["requirements", "what we are looking for", "what we're looking for", "skills", "experience", "must have"], "nice": ["nice to have", "bonus", "preferred"], } def match_heading(s: str): sl = s.lower().strip(":-\x7f ") for key, words in headings.items(): for word in words: if sl == word or sl.startswith(word + " "): return key return None section = None resp_lines = [] req_lines = [] nice_lines = [] for ln in lines: if not ln: continue heading = match_heading(ln) if heading: section = heading continue if section == "responsibilities": resp_lines.append(ln) elif section == "requirements": req_lines.append(ln) elif section == "nice": nice_lines.append(ln) responsibilities = _extract_bullets(resp_lines, max_items=7) requirements = _extract_bullets(req_lines, max_items=7) nice = _extract_bullets(nice_lines, max_items=5) tech_found = [] soft_found = [] low = cleaned.lower() for t in _TECH: if t in low: tech_found.append(t) for s in _SOFT: if s in low: soft_found.append(s) if not responsibilities and not requirements: any_bullets = _extract_bullets(lines, max_items=10) responsibilities = any_bullets[:6] requirements = any_bullets[6:10] focused_parts = [] if responsibilities: focused_parts.append("Responsibilities:\n- " + "\n- ".join(responsibilities)) if requirements: focused_parts.append("Requirements:\n- " + "\n- ".join(requirements)) if nice: focused_parts.append("Nice to have:\n- " + "\n- ".join(nice)) focused_parts.append("Context:\n" + cleaned[:1500]) return { "cleaned": cleaned, "focused_input": "\n\n".join(focused_parts), "responsibilities": responsibilities, "requirements": requirements, "nice": nice, "tech": tech_found, "soft": soft_found, "keywords": _top_keywords(cleaned), } def _model_summarize(text: str, max_length: int, min_length: int) -> str: inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=1024) input_ids = inputs.input_ids.to(device) attention_mask = inputs.attention_mask.to(device) if hasattr(inputs, "attention_mask") else None with torch.no_grad(): outputs = model.generate( input_ids, attention_mask=attention_mask, max_length=max_length, min_length=min_length, num_beams=4, early_stopping=True, ) return tokenizer.decode(outputs[0], skip_special_tokens=True).strip() @app.post("/summarize") async def summarize(req: SummarizeRequest): if req.min_length >= req.max_length: raise HTTPException(status_code=400, detail="min_length must be smaller than max_length.") key = _key(req.text, req.max_length, req.min_length, req.top_skills) if key in cache: return {"summary": cache[key], "cached": True} info = _role_focused_excerpt(req.text) summary = _model_summarize(info["focused_input"], req.max_length, req.min_length) lines = ["Role summary:", summary] if info["requirements"]: lines.append("") lines.append("What the company wants most:") for x in info["requirements"][:7]: lines.append(f"- {x}") if info["responsibilities"]: lines.append("") lines.append("What you would be doing:") for x in info["responsibilities"][:6]: lines.append(f"- {x}") if info["nice"]: lines.append("") lines.append("Nice to have:") for x in info["nice"][:5]: lines.append(f"- {x}") if info["tech"]: uniq = [] for t in _rank_tech_skills(info["tech"]): if t not in uniq: uniq.append(t) lines.append("") lines.append("Top hard skills: " + ", ".join(uniq[: req.top_skills])) if info["soft"]: uniq_soft = [] for s in info["soft"]: if s not in uniq_soft: uniq_soft.append(s) lines.append("") lines.append("Relevant soft skills: " + ", ".join(uniq_soft[:8])) if info["keywords"]: lines.append("") lines.append("Key themes: " + ", ".join(info["keywords"][:6])) lines.append("") lines.append("Interview focus:") if info["requirements"]: for x in info["requirements"][:3]: lines.append(f"- Prepare examples that demonstrate: {x}") elif info["tech"]: for x in _rank_tech_skills(info["tech"])[:3]: lines.append(f"- Be ready to explain your hands-on experience with {x}") else: lines.append("- Prepare examples showing relevant impact, collaboration, and delivery.") out = "\n".join(lines).strip() cache[key] = out return {"summary": out, "cached": False}