import gradio as gr import torch import spaces import logging from deep_translator import GoogleTranslator import pandas as pd from tqdm import tqdm import urllib from bs4 import BeautifulSoup import asyncio from torch.amp import autocast from curl_cffi.requests import AsyncSession from tqdm.asyncio import tqdm from fake_headers import Headers from urllib.parse import urlparse, urlunparse from deep_translator import GoogleTranslator # Limit the number of concurrent workers CONCURRENT_WORKERS = 5 semaphore = asyncio.Semaphore(CONCURRENT_WORKERS) # Configure logging to write messages to a file logging.basicConfig(filename='app.log', level=logging.ERROR) # Configuration max_seq_length = 2048 dtype = None # Auto detection of dtype load_in_4bit = True # Use 4-bit quantization to reduce memory usage # peft_model_name = "limitedonly41/website_qwen2_7b_2" # peft_model_name = "limitedonly41/website_mistral7b_v02" peft_model_name = "unsloth/mistral-7b-instruct-v0.3-bnb-4bit" # Initialize model and tokenizer variables model = None tokenizer = None def get_main_page_url(url): try: # Parse the given URL parsed_url = urlparse(url) # Construct the main page URL (scheme + netloc) print(parsed_url.netloc) main_page_url = urlunparse((parsed_url.scheme, parsed_url.netloc, '', '', '', '')) return main_page_url except Exception as e: return f"Error processing URL: {e}" def translate_text(text): try: text = text[:4990] # Limit the text length to avoid API errors translated_text = GoogleTranslator(source='auto', target='en').translate(text) return translated_text except Exception as e: print(f"An error occurred during translation: {e}") return None async def get_page_bs4(url: str, headers): wrong_result = { 'url': None, 'title': None, 'description': None, 'keywords': None, 'h1': None, 'h2': None, 'h3': None, 'paragraphs': None, 'text': None, 'links': None } async with semaphore: # Limit concurrency async with AsyncSession() as session: wrong_result['url'] = url try: response = await session.get(url, headers=headers, impersonate="chrome", timeout=60, verify=False) except: try: response = await session.get(url, impersonate="chrome", timeout=60, verify=False) except: return wrong_result if response.status_code != 200: return wrong_result soup = BeautifulSoup(response.text, "html.parser") try: title = soup.find('title').text if soup.find('title') else '' except: title = '' try: description = soup.find('meta', attrs={'name': 'description'}) description = description.get("content") if description else '' except: description = '' try: keywords = soup.find('meta', attrs={'name': 'keywords'}) keywords = keywords.get("content") if keywords else '' except: keywords = '' try: h1 = " ".join(h.text for h in soup.find_all('h1')) except: h1 = '' try: h2 = " ".join(h.text for h in soup.find_all('h2')) except: h2 = '' try: h3 = " ".join(h.text for h in soup.find_all('h3')) except: h3 = '' try: paragraphs = " ".join(p.text for p in soup.find_all('p')) except: paragraphs = '' try: menu_tags = [] navs = soup.find_all('nav') uls = soup.find_all('ul') ols = soup.find_all('ol') for tag in navs + uls + ols: menu_tags.extend(tag.find_all('a')) menu_items = [{'text': tag.get_text(strip=True), 'href': tag.get('href')} for tag in menu_tags if tag.get_text(strip=True)] all_menu_texts = ', '.join([item['text'] for item in menu_items]) except: all_menu_texts = '' # all_content = f"{url} {title} {description} {h1} {h2} {h3} {paragraphs}"[:4999] all_content = f" {url} {title} {description} {h1} {h2} {h3} {paragraphs} "[:4999] if len(all_content) < 150: all_content = f" {url} {title} {description} {h1} {h2} {h3} {paragraphs} {all_menu_texts}"[:4999] # all_content = f" {url} {title} {description} {keywords} {h1} {h2} {h3} {paragraphs} "[:4999] # all_content = f" url: {url} title: {title} description: {description} keywords: {keywords} h1: {h1} h2: {h2} h3: {h3} p: {paragraphs} links: {all_menu_texts}"[:4999] result = { 'url': url, 'title': title, 'description': description, 'keywords': keywords, 'h1': h1, 'h2': h2, 'h3': h3, 'paragraphs': paragraphs, 'text': all_content, 'links': all_menu_texts } return result async def main(urls_list): headers_list = [Headers(browser="chrome", os="win").generate() for _ in range(len(urls_list) // 5 + 1)] tasks = [] # Assign headers to each task, rotating every 5 URLs for i, url in enumerate(urls_list): headers = headers_list[i // 5] # Rotate headers every 5 URLs tasks.append(get_page_bs4(url, headers)) # Use tqdm to show progress results = [] for coro in tqdm(asyncio.as_completed(tasks), total=len(tasks)): results.append(await coro) return results def scrape_websites(urls_list): try: import nest_asyncio nest_asyncio.apply() loop = asyncio.get_event_loop() result_data = loop.run_until_complete(main(urls_list)) # print(len(result_data)) except RuntimeError: result_data = asyncio.run(main(urls_list)) return result_data @spaces.GPU() def classify_website(url): from unsloth import FastLanguageModel # Import moved to the top for model loading global model, tokenizer # Declare model and tokenizer as global variables if model is None or tokenizer is None: # Load the model and tokenizer during initialization (in the main process) model, tokenizer = FastLanguageModel.from_pretrained( model_name=peft_model_name, max_seq_length=max_seq_length, dtype=dtype, load_in_4bit=load_in_4bit, ) FastLanguageModel.for_inference(model) # Enable native 2x faster inference main_page_url = get_main_page_url(url) urls = [main_page_url] final_ans_dict = {} print('before scrape_websites') result_data = scrape_websites(urls) data = result_data[0] url = data['url'] text = data['text'] try: if len(text) < 150: # print('Short ', text) prediction = 'Short' final_ans_dict[url] = prediction except: # print(translated) prediction = 'NotScraped' final_ans_dict[url] = prediction translated = translate_text(text) # print(translated) try: if len(translated) < 150: # print(translated) pred = 'Short' return pred except: # print(translated) pred = 'NotScraped' return pred example_input = """https://extensionesdepelo.net/ Hair extensions in Valencia ▶ The best prices for natural hair extensions in Valencia Hair Extensions in Valencia ▶ Professional and Natural ⭐ Hair with more volume and length. Perfect Hair Extensions About us Our works Our salon services Hair extensions Hair removal Reviews of satisfied customers Hair palette colors Contacts Fill out the form Over 7 years of experience in hair extensions, we select the color and texture of hair to match your hair so that the hair extensions look natural Gentle and safe hair extensions so that your hair does not suffer. In a few hours, we will transform rare, weak and short hair into luxurious long and healthy hair. We work exclusively with high-quality hair. Thanks to micro and nano capsules, the extensions will be invisible and comfortable. Free consultation before each extension. We use high-quality hair, time-tested We use small, neat, comfortable capsules and make an unnoticeable transition We consult and answer all questions before and after extensions Safe extensions without discomfort in wearing. Due to the correct placement of the capsules, the result of the extension is invisible.  A procedure that requires the attention and accuracy of the master. With proper hair removal, the structure of native hair is not damaged We provide a large selection of colors Ask the master a question and we will answer all your questions We work in the hot Italian extension technique. This technique is the most comfortable because it does not require much self-care. We recommend doing a correction every 2-3 months. With the Italian technique, you can do various hairstyles and even make ponytails. To form capsules, we use good refractory keratin.  We work with a proven supplier of natural Slavic hair. We have a large selection of colors, lengths and hair structures.""" alpaca_prompt = """Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request. ### Instruction: Describe the topic of website from its text : ### ExampleInput: {} ### ExampleResponse: The website of the master of hair extensions. ### Input: {} ### Response:""" prompt = alpaca_prompt.format(example_input,translated) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") with autocast(device_type='cuda'): inputs = tokenizer(prompt, return_tensors="pt").to(device) outputs = model.generate(**inputs, max_new_tokens=128, use_cache=True) # inputs = tokenizer(prompt, return_tensors="pt").to("cuda") # outputs = model.generate(inputs.input_ids, max_new_tokens=64, use_cache=True) summary = tokenizer.decode(outputs[0], skip_special_tokens=True) final_answer = summary.split("### Response:")[1].strip() return f"{main_page_url}: {final_answer}" # Create a Gradio interface iface = gr.Interface( fn=classify_website, inputs="text", outputs="text", title="Website Topic", description="Enter a URL to get a topic summary of the website content." ) # Launch the interface iface.launch()