Khanishka/AI-document-analyzer-mini

1 2def _detect_doc_type(text: str) -> str:3    text_lower = text.lower()4    opinion_signals = [5        "i believe", "i think", "i feel", "happiness",6        "fear", "love", "hate", "amazing", "terrible",7        "story", "anecdote", "lesson", "mindset",8        "you should", "the best", "the worst"9    ]10    formal_signals = [11        "whereas", "hereby", "pursuant", "therefore",12        "section", "clause", "regulation", "the party",13        "agreement", "terms and conditions"14    ]15    opinion_count = sum(1 for s in opinion_signals if s in text_lower)16    formal_count = sum(1 for s in formal_signals if s in text_lower)17    if formal_count >= 3 and formal_count > opinion_count:18        return "formal"19    return "opinion"20 21import re22 23def _is_noise_amount(value):24    val = value.strip()25    if re.match(r'^97[89][0-9\-]{10,}$', val): return True26    if re.match(r'^[+]?[0-9 ()\-.]{7,}$', val) and len(val) > 8: return True27    return False28 29import re as _re30 31def _clean_text_for_summary(text: str) -> str:32    """Remove noise before sending to AI — emails, phones, URLs, symbols"""33    # Remove emails34    text = _re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '', text)35    # Remove URLs36    text = _re.sub(r'http[s]?://\S+', '', text)37    # Remove phone numbers38    text = _re.sub(r'[\+]?[0-9][\s\-\.]?[(]?[0-9]{3}[)]?[\s\-\.]?[0-9]{3}[\s\-\.]?[0-9]{4,}', '', text)39    # Remove LinkedIn/GitHub URLs40    text = _re.sub(r'(linkedin|github)\.com/\S+', '', text)41    # Remove lines that are just symbols or single words (headers)42    lines = text.split('\n')43    clean_lines = []44    for line in lines:45        stripped = line.strip()46        # Skip very short lines (likely headers/labels)47        if len(stripped) < 15 and stripped.isupper():48            continue49        # Skip lines with mostly special characters50        alpha_ratio = sum(c.isalpha() for c in stripped) / max(len(stripped), 1)51        if alpha_ratio < 0.4 and len(stripped) > 0:52            continue53        clean_lines.append(line)54    return '\n'.join(clean_lines).strip()55 56"""57ai_engine.py58------------59Core AI features powered by Groq LLM + RAG Pipeline:60  - Summarization        (Groq llama-3.3-70b-versatile)61  - Question Answering   (RAG pipeline — TF-IDF retrieval + Groq LLM)62  - Sentiment Analysis   (Groq llama-3.3-70b-versatile)63  - Language Detection   (langdetect)64 65API Key is loaded from environment variable GROQ_API_KEY.66Never hardcode API keys in source files.67"""68 69import os70import re71import time72 73from langdetect import detect, DetectorFactory74from langdetect.lang_detect_exception import LangDetectException75from sklearn.feature_extraction.text import TfidfVectorizer76from sklearn.metrics.pairwise import cosine_similarity77import numpy as np78from groq import Groq79 80DetectorFactory.seed = 4281 82_client = None83 84def _get_client():85    global _client86    if _client is None:87        api_key = os.environ.get("GROQ_API_KEY", "")88        if not api_key:89            raise ValueError(90                "GROQ_API_KEY not set. "91                "Run: import os; os.environ['GROQ_API_KEY'] = 'your-key'"92            )93        _client = Groq(api_key=api_key)94    return _client95 96 97def _ask(prompt: str, max_tokens: int = 400) -> str:98    try:99        r = _get_client().chat.completions.create(100            model="llama-3.3-70b-versatile",101            max_tokens=max_tokens,102            messages=[{"role": "user", "content": prompt}]103        )104        return r.choices[0].message.content.strip()105    except Exception as e:106        print(f"[ai_engine] Groq error: {e}")107        return ""108 109 110def _truncate(text: str, max_chars: int = 4000) -> str:111    return str(text)[:max_chars] if len(str(text)) > max_chars else str(text)112 113 114def _skip_boilerplate(text: str) -> str:115    """Skip copyright/ISBN pages and get to actual content."""116    lines = text.split("\n")117    clean = []118    skip_keywords = ["isbn", "copyright", "all rights reserved", 119                     "published by", "cataloguing", "tel:", "email:",120                     "website:", "harriman", "first published"]121    for line in lines:122        if any(k in line.lower() for k in skip_keywords) and len(line) < 100:123            continue124        clean.append(line)125    result = "\n".join(clean)126    # Skip first 200 chars if they look like metadata127    if result[:200].lower().count("\n") > 5:128        result = result[200:]129    return result.strip()130 131def summarize(text: str) -> dict:132    start = time.time()133    cleaned = _skip_boilerplate(text)134    t = _truncate(cleaned, 4000)135    if len(t.split()) < 30:136        return {"summary": t, "confidence": 1.0, "processing_time_ms": 0}137    out = _ask(138        "Read the document below and write a clear, accurate 3-5 sentence summary.\n"139        "Focus on the main ideas and themes, not metadata or copyright info.\n"140        "Write ONLY the summary sentences. No labels, no bullet points, no preamble.\n\n"141        f"Document:\n{t}", max_tokens=250)142    if not out:143        sentences = re.split(r"(?<=[.!?])\s+", t.strip())144        out = " ".join(sentences[:3])145    confidence = round(min(1.0, len(out.split()) / max(1, len(t.split()) * 0.3)), 2)146    return {"summary": out, "confidence": confidence,147            "processing_time_ms": round((time.time() - start) * 1000, 2)}148 149 150def _split_chunks(text: str, chunk_size: int = 200, overlap: int = 50) -> list:151    words = str(text).split()152    chunks, i = [], 0153    while i < len(words):154        end = min(i + chunk_size, len(words))155        chunks.append(" ".join(words[i:end]))156        if end == len(words): break157        i += chunk_size - overlap158    return chunks159 160 161def _top_chunks(question: str, chunks: list, top_k: int = 5) -> list:162    if not chunks: return chunks[:top_k]163    try:164        corpus = chunks + [question]165        mat = TfidfVectorizer(stop_words="english", ngram_range=(1, 2)).fit_transform(corpus)166        sims = cosine_similarity(mat[-1], mat[:-1]).flatten()167        return [chunks[j] for j in np.argsort(sims)[::-1][:top_k]]168    except Exception:169        return chunks[:top_k]170 171 172def answer_question(context: str, question: str, history: list = None) -> dict:173    start  = time.time()174    chunks = _split_chunks(context)175    tops   = _top_chunks(question, chunks)176    ctx    = " ".join(tops)177    178    # Build conversation history string179    history_str = ""180    if history:181        for turn in history[-4:]:  # last 4 turns max182            history_str += f"User: {turn['question']}\nAssistant: {turn['answer']}\n\n"183    184    prompt = (185        "Answer the question using ONLY the document excerpt below.\n"186        "Give a detailed, specific answer in 3-5 sentences.\n"187        "If the answer is not present say: 'This information is not found in the document.'\n\n"188        f"Document excerpt:\n{ctx[:2500]}\n\n"189    )190    if history_str:191        prompt += f"Previous conversation:\n{history_str}\n"192    prompt += f"Question: {question}\n\nAnswer:"193    194    out = _ask(prompt, max_tokens=350)195    if not out:196        out = "Could not find a specific answer. Try rephrasing your question."197    conf = 0.88 if out and "not found in the document" not in out.lower() else 0.2198    return {"answer": out, "confidence": conf,199            "processing_time_ms": round((time.time() - start) * 1000, 2)}200 201 202def analyze_sentiment(text: str) -> dict:203    start = time.time()204    out = _ask(205        "Analyze the overall sentiment and tone of this text.\n"206        "Rules:\n"207        "- Books about personal growth, wealth, happiness, motivation = POSITIVE\n"208        "- News about disasters, crime, failures = NEGATIVE\n"209        "- Legal documents, technical manuals = NEUTRAL\n"210        "Reply ONLY one word: POSITIVE, NEGATIVE, or NEUTRAL.\n\n"211        f"Text: {_truncate(text, 1500)}", max_tokens=5)212    label = out.strip().upper()213    if label not in ["POSITIVE", "NEGATIVE", "NEUTRAL"]:214        label = "NEUTRAL"215    conf_map = {"POSITIVE": 0.92, "NEGATIVE": 0.91, "NEUTRAL": 0.85}216    return {"label": label, "confidence": conf_map[label],217            "processing_time_ms": round((time.time() - start) * 1000, 2)}218 219 220LANGUAGE_NAMES = {221    "en": "English", "fr": "French", "de": "German", "es": "Spanish",222    "it": "Italian", "pt": "Portuguese", "nl": "Dutch", "ru": "Russian",223    "zh-cn": "Chinese (Simplified)", "zh-tw": "Chinese (Traditional)",224    "ja": "Japanese", "ko": "Korean", "ar": "Arabic", "hi": "Hindi",225    "ta": "Tamil", "te": "Telugu", "ml": "Malayalam", "bn": "Bengali",226    "ur": "Urdu", "tr": "Turkish", "pl": "Polish", "sv": "Swedish",227    "da": "Danish", "fi": "Finnish", "no": "Norwegian", "cs": "Czech",228    "hu": "Hungarian", "ro": "Romanian", "vi": "Vietnamese", "th": "Thai",229    "id": "Indonesian", "ms": "Malay", "el": "Greek", "he": "Hebrew",230    "fa": "Persian", "uk": "Ukrainian", "ca": "Catalan", "sk": "Slovak",231}232 233 234def detect_language(text: str) -> dict:235    start  = time.time()236    sample = str(text)[:1000]237    try:238        lang_code     = detect(sample)239        language_name = LANGUAGE_NAMES.get(lang_code, lang_code.upper())240        confidence    = 0.95241    except LangDetectException:242        lang_code, language_name, confidence = "unknown", "Unknown", 0.0243    return {"language_code": lang_code, "language_name": language_name,244            "confidence": confidence,245            "processing_time_ms": round((time.time() - start) * 1000, 2)}246