Spaces:
Runtime error
Runtime error
Download app.py from zumat/Data-Manipulator: direct link, hf CLI and curl.
- Browser
- Download file 17.2 kB
-
https://huggingface.co/spaces/zumat/Data-Manipulator/resolve/main/app.py
- Command line
-
hf download hf://spaces/zumat/Data-Manipulator/app.py
-
curl -L -o app.py https://huggingface.co/spaces/zumat/Data-Manipulator/resolve/main/app.py
17.2 kB
| import gradio as gr | |
| from email import policy | |
| from email.parser import BytesParser | |
| import base64 | |
| import quopri | |
| import xml.etree.ElementTree as ET | |
| import csv | |
| import os | |
| import json | |
| import gpxpy | |
| import gpxpy.gpx | |
| from icalendar import Calendar | |
| import zipfile | |
| import re | |
| import traceback | |
| from datetime import datetime | |
| from bs4 import BeautifulSoup | |
| def decode_email_content(content, content_transfer_encoding): | |
| if content_transfer_encoding == 'base64': | |
| return base64.b64decode(content) | |
| elif content_transfer_encoding == 'quoted-printable': | |
| return quopri.decodestring(content) | |
| return content | |
| def parse_email(raw_email): | |
| # Parsing the raw email content | |
| msg = BytesParser(policy=policy.default).parsebytes(raw_email) | |
| # Extracting email details | |
| email_details = { | |
| "From": msg["From"], | |
| "To": msg["To"], | |
| "Date": msg["Date"], | |
| "Subject": msg["Subject"], | |
| "Body": "" | |
| } | |
| # Extracting the email body content | |
| if msg.is_multipart(): | |
| for part in msg.iter_parts(): | |
| content_type = part.get_content_type() | |
| content_disposition = part.get("Content-Disposition") | |
| content_transfer_encoding = part.get("Content-Transfer-Encoding") | |
| payload = part.get_payload(decode=False) | |
| if content_type == "text/plain" and content_disposition is None: | |
| body = decode_email_content(payload, content_transfer_encoding) | |
| charset = part.get_content_charset() or 'utf-8' | |
| if isinstance(body, bytes): | |
| body = body.decode(charset) | |
| email_details["Body"] += body.replace('\n', ' ').replace('\r', '') | |
| else: | |
| content_type = msg.get_content_type() | |
| content_transfer_encoding = msg.get("Content-Transfer-Encoding") | |
| payload = msg.get_payload(decode=False) | |
| if content_type == "text/plain": | |
| body = decode_email_content(payload, content_transfer_encoding) | |
| charset = msg.get_content_charset() or 'utf-8' | |
| if isinstance(body, bytes): | |
| body = body.decode(charset) | |
| email_details["Body"] = body.replace('\n', ' ').replace('\r', '') | |
| return email_details | |
| def parse_google_translate_html(filepath): | |
| try: | |
| with open(filepath, 'r', encoding='utf-8') as file: | |
| content = file.read() | |
| soup = BeautifulSoup(content, 'html.parser') | |
| translations = [] | |
| for div in soup.find_all('div', class_='outer-cell mdl-cell mdl-cell--12-col mdl-shadow--2dp'): | |
| title = div.find('p', class_='mdl-typography--title') | |
| if title and "Google Traduttore" in title.text: | |
| content = div.find('div', class_='content-cell mdl-cell mdl-cell--6-col mdl-typography--body-1') | |
| if content: | |
| text = content.text.strip() | |
| link = content.find('a') | |
| if link: | |
| url = link.get('href', '') | |
| query_params = url.split('&') | |
| input_lang = output_lang = input_text = '' | |
| for param in query_params: | |
| if param.startswith('sl='): | |
| input_lang = param.split('=')[1] | |
| elif param.startswith('tl='): | |
| output_lang = param.split('=')[1] | |
| elif param.startswith('q='): | |
| input_text = param.split('=')[1] | |
| input_text = input_text.replace('+', ' ') | |
| language = f"{input_lang} -> {output_lang}" | |
| else: | |
| input_text = '' | |
| language = '' | |
| translation_match = re.search(r'\s+(?:<a.*?>)?"(.*?)"', str(content)) | |
| date_match = re.search(r'(\d{1,2}\s+\w{3}\s+\d{4}),\s+(\d{2}:\d{2}:\d{2}\s+\w+)', text) | |
| if translation_match and date_match: | |
| output_text = translation_match.group(1) | |
| date_str, time_str = date_match.groups() | |
| try: | |
| date = datetime.strptime(f"{date_str} {time_str}", '%d %b %Y %H:%M:%S %Z') | |
| formatted_date = date.strftime('%d/%m/%Y') | |
| formatted_time = date.strftime('%H:%M:%S') | |
| except ValueError: | |
| formatted_date = date_str | |
| formatted_time = time_str | |
| translations.append({ | |
| 'Date': formatted_date, | |
| 'Time': formatted_time, | |
| 'Requested translation': input_text, | |
| 'Language': language | |
| }) | |
| if not translations: | |
| print("No translations found in the file.") | |
| return translations | |
| except Exception as e: | |
| print(f"Error parsing HTML file: {str(e)}") | |
| return [] | |
| def google_translate_to_csv(filepath): | |
| translations = parse_google_translate_html(filepath) | |
| if not translations: | |
| print("No translations to write to CSV.") | |
| return None | |
| output_file = os.path.basename(os.path.splitext(filepath)[0]) + '_converted.csv' | |
| try: | |
| with open(output_file, 'w', newline='', encoding='utf-8') as csvfile: | |
| fieldnames = ['Date', 'Time', 'Requested translation', 'Language'] | |
| writer = csv.DictWriter(csvfile, fieldnames=fieldnames) | |
| writer.writeheader() | |
| for translation in translations: | |
| writer.writerow(translation) | |
| print(f"Successfully wrote {len(translations)} translations to {output_file}") | |
| return output_file | |
| except Exception as e: | |
| print(f"Error writing to CSV file: {str(e)}") | |
| return None | |
| def dataManipulator(converter, filepath): | |
| global output_file | |
| output_file = filepath | |
| try: | |
| if converter == "Email Converter": | |
| records = [] | |
| with open(filepath, 'rb') as file: | |
| raw_email_data = file.read() | |
| email_entries = raw_email_data.split(b'\nFrom ') | |
| for raw_email in email_entries: | |
| if not raw_email.startswith(b'From '): | |
| raw_email = b'From ' + raw_email | |
| parsed_email = parse_email(raw_email) | |
| record_data = { | |
| "date and time": parsed_email["Date"], | |
| "from": parsed_email["From"], | |
| "to": parsed_email["To"], | |
| "Cc": "", # Add Cc if needed, parse_email function does not extract it by default | |
| "Subject": parsed_email["Subject"], | |
| "Content": parsed_email["Body"] | |
| } | |
| records.append(record_data) | |
| output_file = os.path.basename(os.path.splitext(filepath)[0]) + '_converted.csv' | |
| columns = ["date and time", "from", "to", "Cc", "Subject", "Content"] | |
| with open(output_file, 'w', newline='') as csvfile: | |
| writer = csv.DictWriter(csvfile, fieldnames=columns) | |
| writer.writeheader() | |
| for record in records: | |
| writer.writerow(record) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| if converter == "Calendar Converter": | |
| with open(filepath, 'rb') as file: | |
| calendar = Calendar.from_ical(file.read()) | |
| records = [] | |
| for component in calendar.walk(): | |
| if component.name == "VEVENT": | |
| records.append({ | |
| "summary": component.get('summary'), | |
| "dtstart": component.get('dtstart').dt, | |
| "dtend": component.get('dtend').dt, | |
| "description": component.get('description'), | |
| "location": component.get('location') | |
| }) | |
| output_file = os.path.basename(os.path.splitext(filepath)[0]) + '_converted.csv' | |
| with open(output_file, 'w', newline='') as csvfile: | |
| fieldnames = ["summary", "dtstart", "dtend", "description", "location"] | |
| writer = csv.DictWriter(csvfile, fieldnames=fieldnames) | |
| writer.writeheader() | |
| for record in records: | |
| writer.writerow(record) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| if converter == "AppleHealth Converter": | |
| tree = ET.parse(filepath) | |
| root = tree.getroot() | |
| records = [] | |
| columns = ["startDate", "endDate", "device", "sourceName", "type", "unit", "value"] | |
| for record in root.findall('Record'): | |
| record_data = {key: record.attrib.get(key, '') for key in columns} | |
| records.append(record_data) | |
| output_file = os.path.basename(os.path.splitext(filepath)[0]) + '_converted.csv' | |
| with open(output_file, 'w', newline='') as csvfile: | |
| writer = csv.DictWriter(csvfile, fieldnames=columns) | |
| writer.writeheader() | |
| for record in records: | |
| writer.writerow(record) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| if converter == "AppleHealth GPS Converter": | |
| with open(filepath, 'r') as filepath: | |
| gpx = gpxpy.parse(filepath) | |
| records = [] | |
| for track in gpx.tracks: | |
| for segment in track.segments: | |
| for point in segment.points: | |
| records.append({ | |
| "latitude": point.latitude, | |
| "longitude": point.longitude, | |
| "elevation": point.elevation, | |
| "time": point.time | |
| }) | |
| output_file = os.path.basename(os.path.splitext(filepath.name)[0]) + '_converted.csv' | |
| with open(output_file, 'w', newline='') as csvfile: | |
| fieldnames = records[0].keys() | |
| writer = csv.DictWriter(csvfile, fieldnames=fieldnames) | |
| writer.writeheader() | |
| for record in records: | |
| writer.writerow(record) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| if converter == "WhatsApp Converter": | |
| with open(filepath, 'r', encoding='utf-8', errors='ignore') as file: | |
| lines = file.readlines() | |
| records = [] | |
| pattern = r'^\[(\d{2}/\d{2}/\d{2}, \d{2}:\d{2}:\d{2})\] (.+?): (.+)$' | |
| for line in lines: | |
| line = line.strip() | |
| match = re.match(pattern, line) | |
| if match: | |
| date_time, sender, message = match.groups() | |
| date, time = date_time.split(', ') | |
| # Remove any leading special characters from the message | |
| message = re.sub(r'^[\u200E\u200F\u202A-\u202E]', '', message) | |
| records.append({ | |
| "date": date, | |
| "time": time, | |
| "sender": sender, | |
| "message": message | |
| }) | |
| else: | |
| # If the line doesn't match the pattern, it might be a system message or continuation | |
| if records: | |
| records[-1]["message"] += " " + line | |
| output_file = os.path.basename(os.path.splitext(filepath)[0]) + '_converted.csv' | |
| with open(output_file, 'w', newline='', encoding='utf-8') as csvfile: | |
| fieldnames = ["date", "time", "sender", "message"] | |
| writer = csv.DictWriter(csvfile, fieldnames=fieldnames) | |
| writer.writeheader() | |
| for record in records: | |
| writer.writerow(record) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| if converter == "Google Foto Converter": | |
| records = [] | |
| # Unzip the file | |
| with zipfile.ZipFile(filepath, 'r') as zip_ref: | |
| zip_ref.extractall('json_folder') | |
| # Process the extracted files | |
| records = [] | |
| for filename in os.listdir('json_folder'): | |
| if filename.endswith('.json'): | |
| with open(os.path.join('json_folder', filename), 'r') as json_file: | |
| data = json.load(json_file) | |
| records.append(data) | |
| for filename in os.listdir('json_folder'): | |
| if filename.endswith('.json'): | |
| with open(os.path.join('json_folder', filename), 'r') as json_file: | |
| data = json.load(json_file) | |
| records.append(data) | |
| output_file = 'googleFotoConverted.csv' | |
| with open(output_file, 'w', newline='') as csvfile: | |
| fieldnames = set() | |
| for record in records: | |
| fieldnames.update(record.keys()) | |
| writer = csv.DictWriter(csvfile, fieldnames=list(fieldnames)) | |
| writer.writeheader() | |
| for record in records: | |
| writer.writerow(record) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| if converter == "Google Translate Converter": | |
| output_file = google_translate_to_csv(filepath) | |
| return gr.DownloadButton(label=f"Download {output_file}", value=output_file, interactive=True), "" | |
| else: | |
| return gr.DownloadButton(label="Download not ready", value="output_file", interactive=False), "You need to select a converter before" | |
| except Exception as e: | |
| error_message = f"An error occurred: {str(e)}\n\nTraceback:\n{traceback.format_exc()}" | |
| return gr.DownloadButton(label="Download not ready", value="output_file", interactive=False), error_message | |
| demo = gr.Interface( | |
| dataManipulator, | |
| [ | |
| gr.Radio(["Email Converter", "Calendar Converter", "AppleHealth Converter", "AppleHealth GPS Converter", "WhatsApp Converter", "Google Foto Converter", "Google Translate Converter"], label="Select a converter:"), | |
| gr.File(label="Upload a file"), | |
| ], | |
| outputs=[ | |
| gr.DownloadButton(label="Download not ready", value="output_file", interactive=False), | |
| gr.Textbox(label="Error Message", visible=True) | |
| ], | |
| allow_flagging="never", | |
| description=""" | |
| # Data Manipulator | |
| ## This space is designed to convert different datasets into .csv for the workshop [Data Physicalization Practices for Exploiting Personal Data](https://master-interaction-design.notion.site/Data-Physicalization-Practices-for-Exploiting-Personal-Data-0b5b228f0c024c27a66d9bac1c53fd6b?pvs=4). | |
| You can convert the following types of files: | |
| 1. **Email converter**: you can convert the mbox file that your email app can output or can be exported from Google Takeout (if you have a Gmail account) | |
| 2. **Calendar converter**: you can convert the .ics file of any calendar you want | |
| 3. **AppleHealth Converter**: you can convert .xml files from your AppleHealth application | |
| 4. **AppleHealth GPS Converter**: you can convert any .gpx file (tested only with files from AppleHealth application) | |
| 5. **WhatsApp Converter**: you can convert any WhatsApp chat exported as .txt file | |
| 6. **Google Foto Converter**: you can convert multiple JSON files, please upload a .zip file that only contains JSON files to be inserted into a CSV file | |
| 7. **Google Translate Converter**: you can convert a single .html file from Google Takeout -> Google Translate Activity | |
| """, | |
| article=""" | |
| The workshop is sponsored by the [Swiss Design Network](https://swissdesignnetwork.ch/) and the [MAKEAWARE! - Spearhead Project](https://spearhead-amr.github.io/makeaware/) and carried out at the [SUPSI FabLab](https://www.supsi.ch/fablab) in the context of the [Master of Arts in Interaction Design](https://maind.supsi.ch/master-interaction-design/en). | |
| """, | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() |