Spaces:
Running
Running
import os | |
import csv | |
import json | |
import logging | |
import gradio as gr | |
import pandas as pd | |
from tqdm import tqdm | |
import nltk | |
from nltk.tokenize import word_tokenize | |
from nltk.corpus import wordnet | |
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline | |
from huggingface_hub import HfApi, login | |
from datasets import Dataset | |
from datetime import datetime | |
import secrets | |
nltk.download('all') | |
log_dir = "logs" | |
os.makedirs(log_dir, exist_ok=True) | |
logging.basicConfig( | |
filename=os.path.join(log_dir, f"app_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"), | |
level=logging.INFO, | |
format='%(asctime)s - %(levelname)s - %(message)s' | |
) | |
error_dir = "errors" | |
os.makedirs(error_dir, exist_ok=True) | |
error_log_file = os.path.join(error_dir, f"errors_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log") | |
def log_error(error_msg): | |
with open(error_log_file, 'a') as f: | |
f.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - ERROR - {error_msg}\n") | |
HF_TOKEN = os.getenv("HF_TOKEN", secrets.token_hex(16)) | |
login(token=HF_TOKEN) | |
tokenizer = AutoTokenizer.from_pretrained("amd/Instella-3B-Instruct", trust_remote_code=True) | |
model = AutoModelForCausalLM.from_pretrained("amd/Instella-3B-Instruct", trust_remote_code=True) | |
meaning_generator = pipeline("text2text-generation", model="google/flan-t5-large") | |
dataset_dir = "dataset" | |
os.makedirs(dataset_dir, exist_ok=True) | |
csv_file = os.path.join(dataset_dir, "deepfocus_data.csv") | |
parquet_file = os.path.join(dataset_dir, "deepfocus_data.parquet") | |
def process_text_to_csv(input_text): | |
try: | |
tokens = word_tokenize(input_text.lower()) | |
words = list(set(tokens)) | |
data = [] | |
existing_df = pd.read_parquet(parquet_file) if os.path.exists(parquet_file) else pd.DataFrame(columns=["words", "meaning"]) | |
existing_words = set(existing_df["words"].tolist()) | |
for word in tqdm(words, desc="Processing words"): | |
if word in existing_words: | |
continue | |
meanings = [syn.definition() for syn in wordnet.synsets(word)[:3]] or \ | |
[meaning_generator(f"Define the word '{word}'", max_length=100)[0]['generated_text']] | |
data.append({"words": word, "meaning": meanings}) | |
if data: | |
new_df = pd.DataFrame(data) | |
combined_df = pd.concat([existing_df, new_df], ignore_index=True) | |
combined_df.to_parquet(parquet_file, index=False) | |
combined_df.to_csv(csv_file, index=False, encoding='utf-8') | |
return data | |
except Exception as e: | |
log_error(f"Error in process_text_to_csv: {str(e)}") | |
raise | |
def upload_to_huggingface(): | |
try: | |
dataset = Dataset.from_parquet(parquet_file) | |
dataset.push_to_hub("katsukiai/DeepFocus-X3", token=HF_TOKEN) | |
except Exception as e: | |
log_error(f"Error uploading to Hugging Face: {str(e)}") | |
raise | |
def generate_output(input_text): | |
try: | |
data = process_text_to_csv(input_text) | |
upload_to_huggingface() | |
return json.dumps(data, indent=2) | |
except Exception as e: | |
log_error(f"Error in generate_output: {str(e)}") | |
return f"Error: {str(e)}" | |
def view_logs(): | |
try: | |
log_files = os.listdir(log_dir) | |
log_content = "".join(f"\n\n--- {log_file} ---\n\n{open(os.path.join(log_dir, log_file), 'r').read()}" for log_file in log_files) | |
return log_content | |
except Exception as e: | |
log_error(f"Error in view_logs: {str(e)}") | |
return f"Error: {str(e)}" | |
with gr.Blocks(title="DeepFocus-X3") as demo: | |
gr.Markdown("# DeepFocus-X3") | |
with gr.Tabs(): | |
with gr.TabItem("About"): | |
gr.Markdown("## About DeepFocus-X3\nThis application processes text, tokenizes it, extracts unique words, generates meanings, and uploads the dataset to Hugging Face.") | |
with gr.TabItem("Generate all"): | |
input_text = gr.Textbox(label="Input Text", lines=10) | |
output_json = gr.Textbox(label="Output JSON", lines=10) | |
generate_btn = gr.Button("Generate and Upload") | |
generate_btn.click(fn=generate_output, inputs=input_text, outputs=output_json) | |
with gr.TabItem("Logs"): | |
gr.Markdown("## Report using Logs") | |
log_output = gr.Textbox(label="Log Content", lines=20) | |
view_logs_btn = gr.Button("View Logs") | |
view_logs_btn.click(fn=view_logs, inputs=None, outputs=log_output) | |
demo.launch() | |