Add files via upload

src/archive_emails.py

@@ -0,0 +1,152 @@
+import win32com.client
+import sys
+import datetime
+
+
+def list_outlook_mailboxes():
+    """Returns a list of available mailboxes in Outlook."""
+    outlook = win32com.client.Dispatch("Outlook.Application")
+    namespace = outlook.GetNamespace("MAPI")
+    mailboxes = [folder.Name for folder in namespace.Folders]
+    return mailboxes
+
+def find_online_archive(mailboxes, selected_mailbox):
+    """Finds the corresponding Online Archive for the selected mailbox."""
+    archive_name = f"Online Archive - {selected_mailbox}"
+    for mailbox in mailboxes:
+        if mailbox == archive_name:
+            print(f"Mapped Online Archive: {mailbox}")
+            return mailbox
+    print("No Online Archive found for the selected mailbox.")
+    return None
+
+def list_main_folders(mailbox_name):
+    """Lists main folders of the selected mailbox."""
+    outlook = win32com.client.Dispatch("Outlook.Application")
+    namespace = outlook.GetNamespace("MAPI")
+    
+    try:
+        mailbox = namespace.Folders[mailbox_name]
+    except Exception:
+        print(f"Cannot find mailbox: {mailbox_name}. Check the name in Outlook.")
+        return []
+    
+    return [folder.Name for folder in mailbox.Folders]
+
+def main_archive():
+    total_emails_moved = 0
+    total_size_moved_kb = 0
+    """Main function to select a mailbox and folders for archiving."""
+    mailboxes = list_outlook_mailboxes()
+    if not mailboxes:
+        print("No mailboxes found in Outlook.")
+        sys.exit(1)
+
+    print("Available mailboxes:")
+    for idx, mailbox in enumerate(mailboxes, 1):
+        print(f"{idx}. {mailbox}")
+
+    try:
+        choice = int(input("Select mailbox number: ")) - 1
+        if 0 <= choice < len(mailboxes):
+            mailbox_name = mailboxes[choice]
+        else:
+            raise ValueError
+    except ValueError:
+        print("Invalid mailbox selection.")
+        sys.exit(1)
+
+    archive_mailbox = find_online_archive(mailboxes, mailbox_name)
+    if not archive_mailbox:
+        print("No corresponding Online Archive found. Exiting.")
+        sys.exit(1)
+
+    # ✅ Sprawdźmy, czy wybrana skrzynka jest poprawna
+    print(f"Selected mailbox: {mailbox_name}")
+
+    main_folders = list_main_folders(mailbox_name)
+    if not main_folders:
+        print("No main folders found in the selected mailbox.")
+        sys.exit(1)
+
+    print(f"Main folders in '{mailbox_name}':")
+    for idx, folder in enumerate(main_folders, 1):
+        print(f"{idx}. {folder}")
+
+    selected_folders = input("Select folders by number (comma-separated, e.g., 1,3,6): ")
+    try:
+        selected_indices = [int(i.strip()) - 1 for i in selected_folders.split(",")]
+        selected_folder_names = [main_folders[i] for i in selected_indices if 0 <= i < len(main_folders)]
+        print(f"Selected folders: {', '.join(selected_folder_names)}")
+
+        run_type = input("Do you want to perform a dry-run or actually move emails? (dry-run/move): ").strip().lower()
+        if run_type not in ['dry-run', 'move']:
+            print("Invalid selection. Please restart and choose 'dry-run' or 'move'.")
+            sys.exit(1)
+
+        outlook = win32com.client.Dispatch("Outlook.Application")
+        namespace = outlook.GetNamespace("MAPI")
+        mailbox = namespace.Folders[mailbox_name]
+
+        for folder_name in selected_folder_names:
+            try:
+                print(f"Accessing folder: {folder_name}")
+                folder = mailbox.Folders[folder_name]
+
+                try:
+                    archive_folder = namespace.Folders[archive_mailbox].Folders[folder_name]
+                except Exception:
+                    print(f"Archive folder '{folder_name}' does not exist. Attempting to create it...")
+                    try:
+                        archive_folder = namespace.Folders[archive_mailbox].Folders.Add(folder_name)
+                    except Exception as e:
+                        print(f"Failed to create archive folder '{folder_name}'. Skipping. Error: {e}")
+                        continue
+
+                emails_moved, size_moved_kb = process_folder(folder, archive_folder, run_type)
+                total_emails_moved += emails_moved
+                total_size_moved_kb += size_moved_kb
+            except Exception as e:
+               print(f"Error accessing folder {folder_name}: {e}")
+
+        print("Dry-run complete. No emails were moved." if run_type == 'dry-run' else "Email move completed.")
+        print(f"Total emails processed: {total_emails_moved}")
+        print(f"Total size processed: {total_size_moved_kb / 1024:.2f} MB")
+    except (ValueError, IndexError):
+        print("Invalid folder selection.")
+        sys.exit(1)
+
+def process_folder(folder, archive_folder, run_type, depth=0):
+    old_email_count = 0
+    total_size_kb = 0
+    """Processes a folder and its subfolders, counting and optionally moving emails."""
+    old_email_count = 0
+    total_size_kb = 0
+
+    try:
+        # print(f"{'  ' * depth}Processing folder: {folder.Name}")
+        for item in folder.Items:
+            if hasattr(item, 'ReceivedTime') and item.ReceivedTime < (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=365)):
+                old_email_count += 1
+                total_size_kb += item.Size / 1024
+                if run_type == 'move':
+                    item.Move(archive_folder)
+    except Exception as e:
+        print(f"Error processing folder {folder.Name}: {e}")
+
+    # ✅ PRZENIESIONE POZA EXCEPT → teraz zawsze przetwarza podfoldery
+    if old_email_count > 0 :
+        print(f"{'  ' * depth}Folder '{folder.Name}': {old_email_count} emails ({total_size_kb / 1024:.2f} MB) would be moved.")
+
+    for subfolder in folder.Folders:
+        sub_emails, sub_size = process_folder(subfolder, archive_folder, run_type, depth + 1)
+        old_email_count += sub_emails
+        total_size_kb += sub_size
+    return old_email_count, total_size_kb
+
+    process_folder(subfolder, archive_folder, run_type, depth + 1)
+
+
+
+if __name__ == "__main__":
+    main_archive()

src/export_emails.py

@@ -0,0 +1,133 @@
+import sys
+import csv
+import os
+import time
+import win32com.client
+
+
+def list_outlook_mailboxes():
+    """Returns a list of available mailboxes in Outlook."""
+    outlook = win32com.client.Dispatch("Outlook.Application")
+    namespace = outlook.GetNamespace("MAPI")
+    mailboxes = [folder.Name for folder in namespace.Folders]
+    return mailboxes
+
+def get_main_folder_name(base_folder, full_path):
+    """Returns the main folder name from the full path."""
+    full_path = full_path.replace(base_folder + "\\", "")
+    parts = full_path.split("\\")
+    if len(parts) > 0:
+        return parts[0]  # Zwraca główny folder bez adresu e-mail
+    return full_path[:first_slash_pos] if first_slash_pos > 0 else full_path # type: ignore
+
+def process_folder(folder, writer, base_folder, counter, last_log_time):
+    """Recursively scans folders in Outlook and saves email data."""
+    for item in folder.Items:
+        if hasattr(item, 'Class') and item.Class == 43:  # Checks if the item is a MailItem
+            try:
+                writer.writerow([
+                    item.Subject,
+                    item.ReceivedTime.strftime("%Y-%m-%d %H:%M:%S"),
+                    item.Size / 1024,  # Size in KB
+                    item.Size / (1024 * 1024),  # Size in MB
+                    folder.FolderPath.replace(base_folder + "\\", ""),
+                    get_main_folder_name(base_folder, folder.FolderPath)
+                ])
+                counter[0] += 1
+                
+                current_time = time.time()
+                if current_time - last_log_time[0] >= 5:
+                    print(f"Processed {counter[0]} emails...")
+                    last_log_time[0] = current_time
+            except Exception as e:
+                print(f"Error processing email: {e}")
+    
+    for subfolder in folder.Folders:
+        if "PersonMetadata" not in subfolder.FolderPath:
+            process_folder(subfolder, writer, base_folder, counter, last_log_time)
+
+def export_outlook_emails(output_file, root_folder_name):
+    """Exports emails from Outlook to a CSV file."""
+    outlook = win32com.client.Dispatch("Outlook.Application")
+    namespace = outlook.GetNamespace("MAPI")
+    
+    try:
+        root_folder = namespace.Folders[root_folder_name]
+    except Exception as e:
+        print(f"Cannot find folder: {root_folder_name}. Check the name in Outlook.")
+        return
+    
+    os.makedirs(os.path.dirname(output_file), exist_ok=True)  # Tworzy katalog jeśli nie istnieje
+    counter = [0]
+    last_log_time = [time.time()]
+    
+    with open(output_file, mode='w', newline='', encoding='utf-8') as file:
+        writer = csv.writer(file)
+        writer.writerow(["Subject", "Date Received", "Size (KB)", "Size (MB)", "Folder", "Main Folder"])
+        process_folder(root_folder, writer, root_folder.FolderPath, counter, last_log_time)
+    
+    print(f"Export completed. Total emails processed: {counter[0]}. File saved as: {output_file}")
+    print_exported_folder_stats(output_file)
+
+def print_folder_stats(base_folder):
+    """Prints the number of files and total size in main folders."""
+    if not os.path.exists(base_folder):
+        print("No files found in export folder.")
+        return
+    
+    for folder in os.listdir(base_folder):
+        folder_path = os.path.join(base_folder, folder)
+        if os.path.isdir(folder_path):
+            num_files = len([f for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))])
+            total_size = sum(os.path.getsize(os.path.join(folder_path, f)) for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f)))
+            print(f"Folder: {folder} | Files: {num_files} | Total size: {total_size / (1024 * 1024):.2f} MB")
+
+def print_exported_folder_stats(output_file):
+    """Prints the number of emails and total size in exported main folders."""
+    folder_sizes = {}
+    num_files = 0
+    total_size = 0
+    
+    with open(output_file, newline='', encoding='utf-8') as file:
+        reader = csv.reader(file)
+        next(reader)  # Skip header
+        for row in reader:
+            main_folder = row[5]  # Column for Main Folder
+            size_kb = float(row[2])
+            folder_sizes[main_folder] = folder_sizes.get(main_folder, 0) + size_kb
+            num_files += 1
+            total_size += size_kb
+    
+    print("Exported folder statistics:")
+    for folder, size in folder_sizes.items():
+        print(f"Main Folder: {folder} | Emails: {num_files} | Total size: {size / 1024:.2f} MB")
+    print(f"Total exported emails: {num_files} | Overall size: {total_size / 1024:.2f} MB")
+
+def main_export():
+    mailboxes = list_outlook_mailboxes()
+    if not mailboxes:
+        print("No mailboxes found in Outlook.")
+        sys.exit(1)
+
+    for idx, mailbox in enumerate(mailboxes, 1):
+        print(f"{idx}. {mailbox}")
+
+    try:
+        choice = int(input("Select mailbox number: ")) - 1
+        if 0 <= choice < len(mailboxes):
+            folder_name = mailboxes[choice]
+        else:
+            print("Invalid mailbox selection.")
+            sys.exit(1)
+    except (ValueError, IndexError):
+        print("Invalid mailbox selection.")
+        sys.exit(1)
+
+    output_folder = os.path.join(os.getcwd(), "export", folder_name)
+    os.makedirs(output_folder, exist_ok=True)
+    output_csv = os.path.join(output_folder, f"{time.strftime('%Y_%m_%d_%H_%M')}.csv")
+
+    export_outlook_emails(output_csv, folder_name)
+
+if __name__ == "__main__":
+    main_export()

src/main.py

@@ -0,0 +1,29 @@
+import sys
+from export_emails import main_export
+from archive_emails import main_archive  # jeśli dodasz archiwizację
+
+try:
+    import win32com.client
+except ImportError:
+    print("Error: Missing module 'pywin32'. Install it using: pip install pywin32")
+    sys.exit(1)
+
+def main():
+    print("Choose an action:")
+    print("1. Export emails to CSV")
+    print("2. Move old emails to Online Archive")
+    
+    try:
+        choice = int(input("Select an option (1 or 2): "))
+        if choice == 1:
+            main_export()
+        elif choice == 2:
+            main_archive()  # jeśli zaimplementujesz archiwizację
+        else:
+            raise ValueError("Invalid selection.")
+    except ValueError:
+        print("Invalid input. Please select 1 or 2.")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()