Spaces:
Sleeping
Sleeping
| from flask import Flask, render_template, request, redirect, url_for, send_file, flash | |
| import io | |
| import base64 | |
| import matplotlib.pyplot as plt | |
| import preprocessor, helper | |
| import pandas as pd | |
| app = Flask(__name__) | |
| app.secret_key = "change-this-secret" | |
| def figure_to_png_base64(fig): | |
| buf = io.BytesIO() | |
| fig.savefig(buf, format='png', bbox_inches='tight') | |
| plt.close(fig) | |
| buf.seek(0) | |
| return base64.b64encode(buf.read()).decode('utf-8') | |
| def index(): | |
| return render_template('index.html') | |
| def analyze(): | |
| file = request.files.get('chat_file') | |
| if not file or file.filename == '': | |
| flash('Please upload a WhatsApp chat .txt file') | |
| return redirect(url_for('index')) | |
| try: | |
| data = file.read().decode('utf-8') | |
| except Exception: | |
| file.stream.seek(0) | |
| data = file.read().decode('latin-1', errors='ignore') | |
| df = preprocessor.preprocess(data) | |
| users = df['user'].unique().tolist() | |
| if 'group_notification' in users: | |
| users.remove('group_notification') | |
| users.sort() | |
| users.insert(0, 'overall') | |
| selected_user = request.form.get('user') or 'overall' | |
| if selected_user not in users: | |
| selected_user = 'overall' | |
| num_messages, num_words, num_media_messages = helper.fetch_stats(selected_user, df) | |
| charts = {} | |
| if selected_user == 'overall': | |
| x, new_df = helper.most_busy_user(df) | |
| fig, ax = plt.subplots() | |
| ax.bar(x.index, x.values, color=['#ef4444','#10b981','#3b82f6','#f59e0b','#8b5cf6']) | |
| plt.xticks(rotation=45, ha='right') | |
| charts['busy_users'] = figure_to_png_base64(fig) | |
| busy_table = new_df.to_html(classes='table table-striped table-sm', index=False) | |
| else: | |
| busy_table = None | |
| wc_image = helper.create_wordcloud(selected_user, df) | |
| wc_b64 = None | |
| if wc_image is not None: | |
| fig_wc, ax_wc = plt.subplots(figsize=(10, 5)) | |
| ax_wc.imshow(wc_image, interpolation='bilinear') | |
| ax_wc.axis('off') | |
| wc_b64 = figure_to_png_base64(fig_wc) | |
| most_common_df = helper.most_common_words(selected_user, df) | |
| common_b64 = None | |
| if not most_common_df.empty: | |
| fig_mc, ax_mc = plt.subplots() | |
| ax_mc.barh(most_common_df[0], most_common_df[1]) | |
| plt.xticks(rotation=45) | |
| common_b64 = figure_to_png_base64(fig_mc) | |
| emojis_df = helper.emoji_helper(selected_user, df) | |
| emoji_table = emojis_df.to_html(classes='table table-striped table-sm', index=False) | |
| emoji_pie_b64 = None | |
| if not emojis_df.empty: | |
| top_emojis = emojis_df.head(4) | |
| fig_em, ax_em = plt.subplots() | |
| ax_em.pie(top_emojis[1], labels=top_emojis[0], autopct='%0.2f%%') | |
| emoji_pie_b64 = figure_to_png_base64(fig_em) | |
| timeline = helper.monthly_timeline(selected_user, df) | |
| timeline_b64 = None | |
| if not timeline.empty: | |
| fig_tl, ax_tl = plt.subplots() | |
| ax_tl.plot(timeline['time'], timeline['message'], color='#10b981') | |
| plt.xticks(rotation=45, ha='right') | |
| timeline_b64 = figure_to_png_base64(fig_tl) | |
| deltas, rstats = helper.response_time_stats(selected_user, df) | |
| resp_hist_b64 = None | |
| if rstats['count'] > 0: | |
| fig_rt, ax_rt = plt.subplots() | |
| ax_rt.hist(deltas, bins=30, color='#93c5fd', edgecolor='#1f2937') | |
| ax_rt.set_xlabel('Reply time (minutes)') | |
| ax_rt.set_ylabel('Frequency') | |
| ax_rt.set_title('Reply Time Distribution') | |
| resp_hist_b64 = figure_to_png_base64(fig_rt) | |
| request.environ['analysis_df'] = df | |
| request.environ['analysis_selected_user'] = selected_user | |
| return render_template( | |
| 'results.html', | |
| users=users, | |
| selected_user=selected_user, | |
| stats={ | |
| 'messages': num_messages, | |
| 'words': num_words, | |
| 'media': num_media_messages | |
| }, | |
| busy_chart=charts.get('busy_users'), | |
| busy_table=busy_table, | |
| wc_b64=wc_b64, | |
| common_b64=common_b64, | |
| emoji_table=emoji_table, | |
| emoji_pie_b64=emoji_pie_b64, | |
| timeline_b64=timeline_b64, | |
| resp_hist_b64=resp_hist_b64 | |
| ) | |
| def download(): | |
| file = request.files.get('chat_file') | |
| selected_user = request.form.get('user') or 'overall' | |
| if not file or file.filename == '': | |
| flash('Please upload a chat file to generate report') | |
| return redirect(url_for('index')) | |
| try: | |
| data = file.read().decode('utf-8') | |
| except Exception: | |
| file.stream.seek(0) | |
| data = file.read().decode('latin-1', errors='ignore') | |
| df = preprocessor.preprocess(data) | |
| pdf_buffer = helper.generate_pdf_report(selected_user, df) | |
| return send_file( | |
| pdf_buffer, | |
| as_attachment=True, | |
| download_name=f"whatsapp_analysis_{'overall' if selected_user=='overall' else selected_user}.pdf", | |
| mimetype='application/pdf' | |
| ) | |
| if __name__ == '__main__': | |
| app.run(host='0.0.0.0', port=7860, debug=False) | |