netbot/threader_daemon.py at main · Local-Connectivity-Lab/netbot · GitHub

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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

# import asyncio
# import logging
# import signal
# import sys
# from pathlib import Path
# from dotenv import load_dotenv

# # Add netbot-redacted to path
# sys.path.insert(0, str(Path(__file__).parent))

# from threader.imap import Client  # Changed from IMAPClient to Client

# logging.basicConfig(
#     level=logging.INFO,
#     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
# )
# log = logging.getLogger(__name__)

# shutdown_requested = False

# def signal_handler(sig, frame):
#     global shutdown_requested
#     log.info(f"Received signal {sig}, initiating graceful shutdown...")
#     shutdown_requested = True

# async def main():
#     global shutdown_requested

#     # Register signal handlers
#     signal.signal(signal.SIGTERM, signal_handler)
#     signal.signal(signal.SIGINT, signal_handler)

#     log.info("Starting Threader Daemon")
#     log.info("Processes one email completely, then checks for next")

#     # Load environment variables
#     load_dotenv()

#     # Create client once
#     client = Client()  # Changed from IMAPClient()

#     while not shutdown_requested:
#         try:
#             log.info("Checking IMAP for new messages...")

#             # Process emails - this handles ONE email at a time
#             # Returns number of emails processed
#             processed_count = client.synchronize()

#             if processed_count > 0:
#                 log.info(f"Processed {processed_count} email(s)")
#                 # Immediately check for next email (no delay after processing)
#                 continue
#             else:
#                 # No emails found - wait 60 seconds before checking again
#                 log.info("No new emails. Waiting 60 seconds before next check...")
#                 await asyncio.sleep(60)

#         except KeyboardInterrupt:
#             log.info("Keyboard interrupt received")
#             break
#         except Exception as e:
#             log.error(f"Error in main loop: {e}", exc_info=True)
#             # Wait before retrying on error
#             await asyncio.sleep(60)

#     log.info("Threader Daemon stopped")

# if __name__ == "__main__":
#     asyncio.run(main())


#!/usr/bin/env python3
import asyncio
import logging
import signal
import sys
from pathlib import Path
from dotenv import load_dotenv

# Add netbot-redacted to path
sys.path.insert(0, str(Path(__file__).parent))

from threader.imap import Client
from redaction_queue import RedactionQueue

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
log = logging.getLogger(__name__)

shutdown_requested = False

def signal_handler(sig, frame):
    global shutdown_requested
    log.info(f"Received signal {sig}, initiating graceful shutdown...")
    shutdown_requested = True

async def main():
    global shutdown_requested

    # Register signal handlers
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    log.info("Starting Threader Daemon")
    log.info("Processes emails and edit requests sequentially")

    # Load environment variables
    load_dotenv()

    # Create client and queue manager
    client = Client()
    queue = RedactionQueue()

    while not shutdown_requested:
        try:
            # Check for edit jobs in queue FIRST (priority)
            edit_job = queue.get_next_job()

            if edit_job:
                log.info(f"Processing edit job: {edit_job['id']}")
                queue.mark_processing(edit_job['id'])

                try:
                    # Process the edit job
                    client.process_edit_job(edit_job)
                    queue.mark_complete(edit_job['id'])
                    log.info(f"Completed edit job: {edit_job['id']}")
                except Exception as e:
                    log.error(f"Edit job failed: {e}", exc_info=True)
                    queue.mark_failed(edit_job['id'], str(e))

                # Immediately check for next job
                continue

            # No edit jobs, check IMAP
            log.info("Checking IMAP for new messages...")
            processed_count = client.synchronize()

            if processed_count > 0:
                log.info(f"Processed {processed_count} email(s)")
                # Immediately check for next job
                continue
            else:
                # No work to do - wait 60 seconds
                log.info("No pending jobs. Waiting 60 seconds before next check...")
                await asyncio.sleep(60)

        except KeyboardInterrupt:
            log.info("Keyboard interrupt received")
            break
        except Exception as e:
            log.error(f"Error in main loop: {e}", exc_info=True)
            await asyncio.sleep(60)

    log.info("Threader Daemon stopped")

if __name__ == "__main__":
    asyncio.run(main())