Spaces:
Sleeping
Sleeping
File size: 5,422 Bytes
65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 65506c8 fd73a47 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
import os
from PyPDF2 import PdfReader
import pandas as pd
from dotenv import load_dotenv
import json
from datetime import datetime
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
import random
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
class TweetDatasetProcessor:
def __init__(self, fine_tuned_model_name):
load_dotenv()
self.tweets = []
self.personality_profile = {}
self.vectorizer = TfidfVectorizer(stop_words='english')
self.used_tweets = set() # Track used tweets to avoid repetition
# Load fine-tuned model and tokenizer
self.model = AutoModelForCausalLM.from_pretrained(fine_tuned_model_name)
self.tokenizer = AutoTokenizer.from_pretrained(fine_tuned_model_name)
@staticmethod
def _process_line(line):
"""Process a single line."""
line = line.strip()
if not line or line.startswith('http'): # Skip empty lines and URLs
return None
return {
'content': line,
'timestamp': datetime.now(),
'mentions': [word for word in line.split() if word.startswith('@')],
'hashtags': [word for word in line.split() if word.startswith('#')]
}
def extract_text_from_pdf(self, pdf_path):
"""Extract text content from PDF file."""
reader = PdfReader(pdf_path)
text = ""
for page in reader.pages:
text += page.extract_text()
return text
def process_pdf_content(self, text):
"""Process PDF content and clean extracted tweets."""
if not text.strip():
raise ValueError("The uploaded PDF appears to be empty.")
lines = text.split('\n')
clean_tweets = [TweetDatasetProcessor._process_line(line) for line in lines]
self.tweets = [tweet for tweet in clean_tweets if tweet]
if not self.tweets:
raise ValueError("No tweets were extracted from the PDF. Ensure the content is properly formatted.")
# Save the processed tweets to a CSV
df = pd.DataFrame(self.tweets)
df.to_csv('processed_tweets.csv', index=False)
return df
def categorize_tweets(self):
"""Cluster tweets into categories using KMeans."""
all_tweets = [tweet['content'] for tweet in self.tweets]
if not all_tweets:
raise ValueError("No tweets available for clustering.")
tfidf_matrix = self.vectorizer.fit_transform(all_tweets)
kmeans = KMeans(n_clusters=5, random_state=1)
kmeans.fit(tfidf_matrix)
for i, tweet in enumerate(self.tweets):
tweet['category'] = f"Category {kmeans.labels_[i]}"
return pd.DataFrame(self.tweets)
def analyze_personality(self, max_tweets=50):
"""Comprehensive personality analysis using a limited subset of tweets."""
if not self.tweets:
raise ValueError("No tweets available for personality analysis.")
all_tweets = [tweet['content'] for tweet in self.tweets][:max_tweets]
analysis_prompt = f"""Perform a deep psychological analysis of the author based on these tweets:
Core beliefs, emotional tendencies, cognitive patterns, etc.
Tweets for analysis:
{json.dumps(all_tweets, indent=2)}
"""
input_ids = self.tokenizer.encode(analysis_prompt, return_tensors='pt')
output = self.model.generate(input_ids, max_length=500, num_return_sequences=1, temperature=0.7)
personality_analysis = self.tokenizer.decode(output[0], skip_special_tokens=True)
self.personality_profile = personality_analysis
return self.personality_profile
def generate_tweet(self, context="", sample_size=3):
"""Generate a new tweet by sampling random tweets and avoiding repetition."""
if not self.tweets:
return "Error: No tweets available for generation."
# Randomly sample unique tweets
available_tweets = [tweet for tweet in self.tweets if tweet['content'] not in self.used_tweets]
if len(available_tweets) < sample_size:
self.used_tweets.clear() # Reset used tweets if all have been used
available_tweets = self.tweets
sampled_tweets = random.sample(available_tweets, sample_size)
sampled_contents = [tweet['content'] for tweet in sampled_tweets]
# Update the used tweets tracker
self.used_tweets.update(sampled_contents)
# Truncate personality profile to avoid token overflow
personality_profile_excerpt = self.personality_profile[:400] if len(self.personality_profile) > 400 else self.personality_profile
# Construct the prompt
prompt = f"""Based on this personality profile:
{personality_profile_excerpt}
Current context or topic (if any):
{context}
Tweets for context:
{', '.join(sampled_contents)}
**Only generate the tweet. Do not include analysis, explanation, or any other content.**
"""
input_ids = self.tokenizer.encode(prompt, return_tensors='pt')
output = self.model.generate(input_ids, max_length=150, num_return_sequences=1, temperature=1.0)
generated_tweet = self.tokenizer.decode(output[0], skip_special_tokens=True).strip()
return generated_tweet
|