from flask import Flask, render_template, request, jsonify, send_file import json import re from duckduckgo_search import DDGS import requests from bs4 import BeautifulSoup import fitz # PyMuPDF import urllib3 import pandas as pd import io import ast from groq import Groq import os app = Flask(__name__) search_prompt = """ The user will provide a detailed description of a technical problem they are trying to solve in the context of intellectual property (IP) and patents. Your task is to generate some (2 to 5) highly specific and relevant search queries for Google, aimed at finding research papers closely related to the user's problem. Each search query should: 1. Be crafted to find research papers, articles, or academic resources that address similar issues or solutions. 2. Be focused and precise, avoiding generic or overly broad terms. Provide the search queries in the following **JSON format**. There should be no extra text, only the search queries as values. **Example Output:** ```json { "1": "user authentication 5G cryptographic keys identity management", "2": "5G authentication security issues cryptography 3GPP key management" } ``` """ infringement_prompt = """You are an expert assistant designed to evaluate the novelty and inventiveness of patents by comparing them with existing documents. Your task is to analyze the background of a given patent and the first page of a related document to determine how well the document covers the problems mentioned in the patent. # Instructions: Understand the Patent Background: Carefully read and comprehend the background information provided for the patent. Identify the key problems that the patent aims to address. Analyze the Document: Review the provided document. Focus on identifying any problems that are similar to those mentioned in the patent background. Evaluate Coverage: Assess how well the document covers the problems mentioned in the patent. Use the following scoring system: Score 5: The document explicitly discusses the same problems as the patent, indicating that the problems are not novel. Score 4: The document discusses problems that are very similar to those in the patent, significantly impacting the novelty of the patent's problems. Score 3: The document mentions problems that are somewhat similar to those in the patent, but the coverage is not extensive enough to fully block the novelty of the patent's problems. Score 2: The document mentions problems that are similar in some ways but are clearly different from those in the patent. Score 1: The document touches upon related problems but does not directly address the specific problems mentioned in the patent. Score 0: The document does not discuss any problems related to those in the patent. Provide a Score: Based on your analysis, provide a score from 0 to 5 indicating how well the document covers the problems mentioned in the patent. Justify Your Score: Briefly explain the reasoning behind your score, highlighting specific similarities or differences between the problems discussed in the patent and the document. # Output Format: No details or explainations are required, just the results in the required **JSON** format with no additional word. { 'score': [Your Score], 'justification': "[Your Justification]" } """ def ask_ollama(user_message, model='llama-3.3-70b-versatile', system_prompt=search_prompt): client = Groq(api_key=os.environ.get("GROQ_API_KEY")) response = client.chat.completions.create( model=model, messages=[ { "role": "system", "content": system_prompt }, { "role": "user", "content": user_message } ], stream=False, ) ai_reply = response.choices[0].message.content print(f"AI REPLY json:\n{ai_reply}") # Process the response to ensure we return valid JSON try: # First, try to parse it directly in case it's already valid JSON print(f"AI REPLY:\n{ai_reply}") return ast.literal_eval(ai_reply.replace('json\n', '').replace('```', '')) except Exception as e: print(f"ERROR:\n{e}") # If it's not valid JSON, try to extract JSON from the text return { "1": "Error parsing response. Please try again.", "2": "Error parsing response. Please try again." } def search_web(topic, max_references=5, data_type="pdf"): """Search the web using DuckDuckGo and return results.""" doc_list = [] with DDGS(verify=False) as ddgs: i = 0 for r in ddgs.text(topic, region='wt-wt', safesearch='On', timelimit='n'): if i >= max_references: break doc_list.append({"type": data_type, "title": r['title'], "body": r['body'], "url": r['href']}) i += 1 return doc_list def analyze_pdf_novelty(patent_background, url, data_type="pdf"): """Extract first page text from PDF or background from patent and evaluate novelty""" try: # Disable SSL warnings urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Extract text based on the type if data_type == "pdf": headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Accept": "application/pdf" } response = requests.get(url, headers=headers, timeout=20, verify=False) if response.status_code != 200: print(f"Failed to download PDF (status code: {response.status_code})") return {"error": f"Failed to download PDF (status code: {response.status_code})"} # Extract first page text try: pdf_document = fitz.open(stream=response.content, filetype="pdf") if pdf_document.page_count == 0: return {"error": "PDF has no pages"} first_page = pdf_document.load_page(0) text = first_page.get_text() except Exception as e: return {"error": f"Error processing PDF: {str(e)}"} elif data_type == "patent": # Extract background from patent print("extract from patent") try: response = requests.get(url, timeout=20, verify=False) if response.status_code != 200: print(f"Failed to access patent (status code: {response.status_code})") return {"error": f"Failed to access patent (status code: {response.status_code})"} content = response.content.decode('utf-8').replace("\n", "") soup = BeautifulSoup(content, 'html.parser') section = soup.find('section', itemprop='description', itemscope='') matches = re.findall(r"background(.*?)(?:summary|description of the drawing)", str(section), re.DOTALL | re.IGNORECASE) if matches: text = BeautifulSoup(matches[0], "html.parser").get_text(separator=" ").strip() else: text = "Background section not found in patent." except Exception as e: return {"error": f"Error processing patent: {str(e)}"} elif data_type == "web": try: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Accept": "application/pdf" } response = requests.get(url, headers=headers, timeout=20, verify=False) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') full_text = soup.get_text() text = re.sub(r'\n+', ' ', full_text)[:5000] except requests.RequestException as e: return {"error": f"Error fetching the page: {str(e)}"} else: return {"error": "Unknown document type"} # Analyze with Ollama result = ask_ollama( user_message=f"Patent background:\n{patent_background}\n\nDocument first page:\n{text}", system_prompt=infringement_prompt ) return result except Exception as e: return {"error": f"Error: {str(e)}"} @app.route('/') def home(): return render_template('index.html') @app.route('/chat', methods=['POST']) def chat(): user_message = request.form.get('message') ai_reply = ask_ollama(user_message) return jsonify({'reply': ai_reply}) @app.route('/search', methods=['POST']) def search(): query = request.form.get('query') pdf_checked = request.form.get('pdfOption') == 'true' patent_checked = request.form.get('patentOption') == 'true' web_checked = request.form.get('webOption') == 'true' or request.form.get('webOption') == 'on' if not query: return jsonify({'error': 'No query provided', 'results': []}) all_results = [] try: # Handle various combinations if pdf_checked: pdf_query = f"{query} filetype:pdf" pdf_results = search_web(pdf_query, max_references=5, data_type="pdf") all_results.extend(pdf_results) if patent_checked: patent_query = f"{query} site:patents.google.com" patent_results = search_web(patent_query, max_references=5, data_type="patent") all_results.extend(patent_results) if web_checked: # For web, we don't add anything to the query web_results = search_web(query, max_references=5, data_type="web") all_results.extend(web_results) # If nothing is checked, default to web search if not (pdf_checked or patent_checked or web_checked): web_results = search_web(query, max_references=5, data_type="web") all_results.extend(web_results) return jsonify({'results': all_results}) except Exception as e: print(f"Error performing search: {e}") return jsonify({'error': str(e), 'results': []}) @app.route('/analyze', methods=['POST']) def analyze(): data = request.json if not data or 'patent_background' not in data or 'pdf_url' not in data: return jsonify({'error': 'Missing required parameters', 'result': None}) try: patent_background = data['patent_background'] url = data['pdf_url'] data_type = data.get('data_type', 'pdf') # Default to pdf if not specified result = analyze_pdf_novelty(patent_background, url, data_type) return jsonify({'result': result}) except Exception as e: print(f"Error analyzing document: {e}") return jsonify({'error': str(e), 'result': None}) @app.route('/export-excel', methods=['POST']) def export_excel(): try: data = request.json if not data or 'tableData' not in data: return jsonify({'error': 'No table data provided'}) # Create pandas DataFrame from the data df = pd.DataFrame(data['tableData']) # Get the user query user_query = data.get('userQuery', 'No query provided') # Create a BytesIO object to store the Excel file output = io.BytesIO() # Create Excel file with xlsxwriter engine with pd.ExcelWriter(output, engine='xlsxwriter') as writer: # Write the data to a sheet named 'Results' df.to_excel(writer, sheet_name='Results', index=False) # Get workbook and worksheet objects workbook = writer.book worksheet = writer.sheets['Results'] # Add a sheet for the query query_sheet = workbook.add_worksheet('Query') query_sheet.write(0, 0, 'Patent Query') query_sheet.write(1, 0, user_query) # Adjust column widths for i, col in enumerate(df.columns): # Get maximum column width max_len = max( df[col].astype(str).map(len).max(), len(col) ) + 2 # Set column width (limit to 100 to avoid issues) worksheet.set_column(i, i, min(max_len, 100)) # Seek to the beginning of the BytesIO object output.seek(0) # Return the Excel file return send_file( output, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', as_attachment=True, download_name='patent_search_results.xlsx' ) except Exception as e: print(f"Error exporting Excel: {e}") return jsonify({'error': str(e)}) if __name__ == '__main__': app.run(host="0.0.0.0", port=7860)