app.getemail.send_new_user_notification() - Code Metrics - Inspection of "Add email notification to global admin when new us..." - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( cd6825...32c582 )

by William

created 2025-12-11 20:51 UTC

app.getemail.send_new_user_notification() B

↳ Parent: app.getemail

Complexity

Conditions

Size

Total Lines	114
Code Lines	47

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	47
nop	2
dl	0
loc	114
rs	7.8012
c	0
b	0
f	0

How to fix Long Method

#!/usr/bin/env python3
"""
Email balance import script for pycashflow.
Processes IMAP email inboxes to extract and import balance information.

This script can be run standalone from cron or called from within the application.
"""
import os
import sys

# When run as standalone script, add parent directory to path for imports
# This allows 'from app import ...' to work when called from cron
parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
if parent_dir not in sys.path:
    sys.path.insert(0, parent_dir)

import imaplib
import email
from email.header import decode_header
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
from datetime import datetime, timedelta
from app import db
from app.models import User, Email, Balance


def process_email_balances():
    """
    Process email inboxes for all users with email settings configured
    and update their balance information from bank emails.

    This function supports both SQLite and PostgreSQL through SQLAlchemy.
    """
    # OUTER LOOP: Get all users with email settings configured
    users_with_email = db.session.query(User, Email).join(
        Email, User.id == Email.user_id
    ).all()

    for user, email_config in users_with_email:
        user_id = user.id
        username = email_config.email
        password = email_config.password
        imap_server = email_config.server
        subjectstr = email_config.subjectstr
        startstr = email_config.startstr
        endstr = email_config.endstr

        try:
            # Create IMAP connection for THIS user
            imap = imaplib.IMAP4_SSL(imap_server)
            imap.login(username, password)

            status, messages = imap.select("INBOX", readonly=True)

            # Filter for emails from the past day to limit inbox processing
            # IMAP SINCE searches for messages with Date on or after the specified date
            yesterday = (datetime.now() - timedelta(days=1)).strftime("%d-%b-%Y")
            status, message_ids = imap.search(None, f'SINCE {yesterday}')

            # Parse message IDs from the search result
            if message_ids[0]:
                email_ids = message_ids[0].split()
            else:
                email_ids = []

            print(f"Found {len(email_ids)} email(s) from the past day for user {user_id}")

            email_content = {}

            # INNER LOOP: Process only emails from the past day
            for email_id in email_ids:
                try:
                    res, msg = imap.fetch(email_id, "(RFC822)")
                    for response in msg:
                        if isinstance(response, tuple):
                            msg = email.message_from_bytes(response[1])

                            # Decode subject
                            subject, encoding = decode_header(msg["Subject"])[0]
                            if isinstance(subject, bytes):
                                try:
                                    subject = subject.decode(encoding)
                                except:
                                    subject = "subject"

                            # Decode sender
                            From, encoding = decode_header(msg.get("From"))[0]
                            if isinstance(From, bytes):
                                try:
                                    From = From.decode(encoding)
                                except:
                                    pass

                            # Extract email body
                            if msg.is_multipart():
                                for part in msg.walk():
                                    content_type = part.get_content_type()
                                    content_disposition = str(part.get("Content-Disposition"))
                                    try:
                                        body = part.get_payload(decode=True).decode()
                                    except:
                                        pass
                                    if content_type == "text/plain" and "attachment" not in content_disposition:
                                        try:
                                            email_content[subject] = body
                                        except:
                                            pass
                            else:
                                content_type = msg.get_content_type()
                                body = msg.get_payload(decode=True).decode()
                                if content_type == "text/plain":
                                    try:
                                        email_content[subject] = body
                                    except:
                                        pass
                except:
                    pass  # Skip individual email errors

            # Extract balance from emails for THIS user
            try:
                start_index = email_content[subjectstr].find(startstr) + len(startstr)
                end_index = email_content[subjectstr].find(endstr)
                new_balance = email_content[subjectstr][start_index:end_index].replace(',', '')
                new_balance = new_balance.replace('$', '')
                new_balance = float(new_balance)

                # Insert balance WITH user_id using SQLAlchemy ORM
                balance = Balance(
                    amount=new_balance,
                    date=datetime.today().date(),
                    user_id=user_id
                )
                db.session.add(balance)
                db.session.commit()
                print(f"Successfully imported balance ${new_balance} for user {user_id}")
            except KeyError:
                # No email with the specified subject found
                pass
            except Exception as e:
                # No balance found in emails for this user
                print(f"Could not extract balance for user {user_id}: {e}")

            # Close IMAP connection for THIS user
            imap.close()
            imap.logout()

        except Exception as e:
            # Failed to connect to IMAP for this user - skip to next user
            print(f"Failed to process emails for user {user_id}: {e}")
            continue


def send_new_user_notification(new_user_name, new_user_email):
    """
    Send an email notification to the global admin when a new user registers.

    Args:
        new_user_name: Name of the newly registered user
        new_user_email: Email of the newly registered user

    Returns:
        bool: True if email was sent successfully, False otherwise
    """
    try:
        # Get the global admin user
        global_admin = User.query.filter_by(is_global_admin=True).first()

        if not global_admin:
            print("No global admin found, skipping notification")
            return False

        # Get email settings for the global admin
        email_config = Email.query.filter_by(user_id=global_admin.id).first()

        if not email_config:
            print(f"No email settings configured for global admin {global_admin.id}, skipping notification")
            return False

        # Extract email credentials
        from_email = email_config.email
        password = email_config.password
        imap_server = email_config.server

        # Convert IMAP server to SMTP server
        # Common patterns: imap.gmail.com -> smtp.gmail.com, imap.mail.yahoo.com -> smtp.mail.yahoo.com
        smtp_server = imap_server.replace('imap', 'smtp')

        # Create the email message
        msg = MIMEMultipart('alternative')
        msg['From'] = from_email
        msg['To'] = from_email  # Send to the global admin's own email
        msg['Subject'] = 'New User Registration - PyCashFlow'

        # Create plain text version
        text_content = f"""
A new user has registered on PyCashFlow.

Name: {new_user_name}
Email: {new_user_email}
Registration Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

This user account is currently inactive and requires approval before they can log in.
Please log in to your PyCashFlow account to activate this user.
"""

        # Create HTML version
        html_content = f"""
<html>
  <body>
    <h2>New User Registration</h2>
    <p>A new user has registered on PyCashFlow.</p>
    <table style="border-collapse: collapse; margin-top: 10px;">
      <tr>
        <td style="padding: 5px; font-weight: bold;">Name:</td>
        <td style="padding: 5px;">{new_user_name}</td>
      </tr>
      <tr>
        <td style="padding: 5px; font-weight: bold;">Email:</td>
        <td style="padding: 5px;">{new_user_email}</td>
      </tr>
      <tr>
        <td style="padding: 5px; font-weight: bold;">Registration Date:</td>
        <td style="padding: 5px;">{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</td>
      </tr>
    </table>
    <p style="margin-top: 15px;">
      <strong>Note:</strong> This user account is currently inactive and requires approval before they can log in.
      Please log in to your PyCashFlow account to activate this user.
    </p>
  </body>
</html>
"""

        # Attach both versions
        part1 = MIMEText(text_content, 'plain')
        part2 = MIMEText(html_content, 'html')
        msg.attach(part1)
        msg.attach(part2)

        # Send the email using SMTP
        # Try port 587 with STARTTLS first (most common)
        try:
            server = smtplib.SMTP(smtp_server, 587, timeout=10)
            server.starttls()
            server.login(from_email, password)
            server.sendmail(from_email, from_email, msg.as_string())
            server.quit()
            print(f"Successfully sent new user notification email to {from_email}")
            return True
        except Exception as e:
            # If port 587 fails, try port 465 with SSL
            print(f"Failed to send via port 587, trying SSL on port 465: {e}")
            try:
                server = smtplib.SMTP_SSL(smtp_server, 465, timeout=10)
                server.login(from_email, password)
                server.sendmail(from_email, from_email, msg.as_string())
                server.quit()
                print(f"Successfully sent new user notification email to {from_email} via SSL")
                return True
            except Exception as e2:
                print(f"Failed to send email via both ports: {e2}")
                return False

    except Exception as e:
        print(f"Error sending new user notification: {e}")
        return False


# Allow script to be run standalone from cron
if __name__ == "__main__":
    from app import create_app

    print("Starting email balance import...")
    app = create_app()

    with app.app_context():
        try:
            process_email_balances()
            print("Email balance import completed successfully")
        except Exception as e:
            print(f"Email balance import failed: {e}")
            sys.exit(1)


1			#!/usr/bin/env python3
2			"""
3			Email balance import script for pycashflow.
4			Processes IMAP email inboxes to extract and import balance information.
5
6			This script can be run standalone from cron or called from within the application.
7			"""
8			import os
9			import sys
10
11			# When run as standalone script, add parent directory to path for imports
12			# This allows 'from app import ...' to work when called from cron
13			parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
14			if parent_dir not in sys.path:
15			sys.path.insert(0, parent_dir)
16
17			import imaplib
18			import email
19			from email.header import decode_header
20			from email.mime.text import MIMEText
21			from email.mime.multipart import MIMEMultipart
22			import smtplib
23			from datetime import datetime, timedelta
24			from app import db
25			from app.models import User, Email, Balance
26
27
28			def process_email_balances():
29			"""
30			Process email inboxes for all users with email settings configured
31			and update their balance information from bank emails.
32
33			This function supports both SQLite and PostgreSQL through SQLAlchemy.
34			"""
35			# OUTER LOOP: Get all users with email settings configured
36			users_with_email = db.session.query(User, Email).join(
37			Email, User.id == Email.user_id
38			).all()
39
40			for user, email_config in users_with_email:
41			user_id = user.id
42			username = email_config.email
43			password = email_config.password
44			imap_server = email_config.server
45			subjectstr = email_config.subjectstr
46			startstr = email_config.startstr
47			endstr = email_config.endstr
48
49			try:
50			# Create IMAP connection for THIS user
51			imap = imaplib.IMAP4_SSL(imap_server)
52			imap.login(username, password)
53
54			status, messages = imap.select("INBOX", readonly=True)
55
56			# Filter for emails from the past day to limit inbox processing
57			# IMAP SINCE searches for messages with Date on or after the specified date
58			yesterday = (datetime.now() - timedelta(days=1)).strftime("%d-%b-%Y")
59			status, message_ids = imap.search(None, f'SINCE {yesterday}')
60
61			# Parse message IDs from the search result
62			if message_ids[0]:
63			email_ids = message_ids[0].split()
64			else:
65			email_ids = []
66
67			print(f"Found {len(email_ids)} email(s) from the past day for user {user_id}")
68
69			email_content = {}
70
71			# INNER LOOP: Process only emails from the past day
72			for email_id in email_ids:
73			try:
74			res, msg = imap.fetch(email_id, "(RFC822)")
75			for response in msg:
76			if isinstance(response, tuple):
77			msg = email.message_from_bytes(response[1])
78
79			# Decode subject
80			subject, encoding = decode_header(msg["Subject"])[0]
81			if isinstance(subject, bytes):
82			try:
83			subject = subject.decode(encoding)
84			except:
85			subject = "subject"
86
87			# Decode sender
88			From, encoding = decode_header(msg.get("From"))[0]
89			if isinstance(From, bytes):
90			try:
91			From = From.decode(encoding)
92			except:
93			pass
94
95			# Extract email body
96			if msg.is_multipart():
97			for part in msg.walk():
98			content_type = part.get_content_type()
99			content_disposition = str(part.get("Content-Disposition"))
100			try:
101			body = part.get_payload(decode=True).decode()
102			except:
103			pass
104			if content_type == "text/plain" and "attachment" not in content_disposition:
105			try:
106			email_content[subject] = body
107			except:
108			pass
109			else:
110			content_type = msg.get_content_type()
111			body = msg.get_payload(decode=True).decode()
112			if content_type == "text/plain":
113			try:
114			email_content[subject] = body
115			except:
116			pass
117			except:
118			pass # Skip individual email errors
119
120			# Extract balance from emails for THIS user
121			try:
122			start_index = email_content[subjectstr].find(startstr) + len(startstr)
123			end_index = email_content[subjectstr].find(endstr)
124			new_balance = email_content[subjectstr][start_index:end_index].replace(',', '')
125			new_balance = new_balance.replace('$', '')
126			new_balance = float(new_balance)
127
128			# Insert balance WITH user_id using SQLAlchemy ORM
129			balance = Balance(
130			amount=new_balance,
131			date=datetime.today().date(),
132			user_id=user_id
133			)
134			db.session.add(balance)
135			db.session.commit()
136			print(f"Successfully imported balance ${new_balance} for user {user_id}")
137			except KeyError:
138			# No email with the specified subject found
139			pass
140			except Exception as e:
141			# No balance found in emails for this user
142			print(f"Could not extract balance for user {user_id}: {e}")
143
144			# Close IMAP connection for THIS user
145			imap.close()
146			imap.logout()
147
148			except Exception as e:
149			# Failed to connect to IMAP for this user - skip to next user
150			print(f"Failed to process emails for user {user_id}: {e}")
151			continue
152
153
154			def send_new_user_notification(new_user_name, new_user_email):
155			"""
156			Send an email notification to the global admin when a new user registers.
157
158			Args:
159			new_user_name: Name of the newly registered user
160			new_user_email: Email of the newly registered user
161
162			Returns:
163			bool: True if email was sent successfully, False otherwise
164			"""
165			try:
166			# Get the global admin user
167			global_admin = User.query.filter_by(is_global_admin=True).first()
168
169			if not global_admin:
170			print("No global admin found, skipping notification")
171			return False
172
173			# Get email settings for the global admin
174			email_config = Email.query.filter_by(user_id=global_admin.id).first()
175
176			if not email_config:
177			print(f"No email settings configured for global admin {global_admin.id}, skipping notification")
178			return False
179
180			# Extract email credentials
181			from_email = email_config.email
182			password = email_config.password
183			imap_server = email_config.server
184
185			# Convert IMAP server to SMTP server
186			# Common patterns: imap.gmail.com -> smtp.gmail.com, imap.mail.yahoo.com -> smtp.mail.yahoo.com
187			smtp_server = imap_server.replace('imap', 'smtp')
188
189			# Create the email message
190			msg = MIMEMultipart('alternative')
191			msg['From'] = from_email
192			msg['To'] = from_email # Send to the global admin's own email
193			msg['Subject'] = 'New User Registration - PyCashFlow'
194
195			# Create plain text version
196			text_content = f"""
197			A new user has registered on PyCashFlow.
198
199			Name: {new_user_name}
200			Email: {new_user_email}
201			Registration Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
202
203			This user account is currently inactive and requires approval before they can log in.
204			Please log in to your PyCashFlow account to activate this user.
205			"""
206
207			# Create HTML version
208			html_content = f"""
209			<html>
210			<body>
211			<h2>New User Registration</h2>
212			<p>A new user has registered on PyCashFlow.</p>
213			<table style="border-collapse: collapse; margin-top: 10px;">
214			<tr>
215			<td style="padding: 5px; font-weight: bold;">Name:</td>
216			<td style="padding: 5px;">{new_user_name}</td>
217			</tr>
218			<tr>
219			<td style="padding: 5px; font-weight: bold;">Email:</td>
220			<td style="padding: 5px;">{new_user_email}</td>
221			</tr>
222			<tr>
223			<td style="padding: 5px; font-weight: bold;">Registration Date:</td>
224			<td style="padding: 5px;">{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</td>
225			</tr>
226			</table>
227			<p style="margin-top: 15px;">
228			<strong>Note:</strong> This user account is currently inactive and requires approval before they can log in.
229			Please log in to your PyCashFlow account to activate this user.
230			</p>
231			</body>
232			</html>
233			"""
234
235			# Attach both versions
236			part1 = MIMEText(text_content, 'plain')
237			part2 = MIMEText(html_content, 'html')
238			msg.attach(part1)
239			msg.attach(part2)
240
241			# Send the email using SMTP
242			# Try port 587 with STARTTLS first (most common)
243			try:
244			server = smtplib.SMTP(smtp_server, 587, timeout=10)
245			server.starttls()
246			server.login(from_email, password)
247			server.sendmail(from_email, from_email, msg.as_string())
248			server.quit()
249			print(f"Successfully sent new user notification email to {from_email}")
250			return True
251			except Exception as e:
252			# If port 587 fails, try port 465 with SSL
253			print(f"Failed to send via port 587, trying SSL on port 465: {e}")
254			try:
255			server = smtplib.SMTP_SSL(smtp_server, 465, timeout=10)
256			server.login(from_email, password)
257			server.sendmail(from_email, from_email, msg.as_string())
258			server.quit()
259			print(f"Successfully sent new user notification email to {from_email} via SSL")
260			return True
261			except Exception as e2:
262			print(f"Failed to send email via both ports: {e2}")
263			return False
264
265			except Exception as e:
266			print(f"Error sending new user notification: {e}")
267			return False
268
269
270			# Allow script to be run standalone from cron
271			if __name__ == "__main__":
272			from app import create_app
273
274			print("Starting email balance import...")
275			app = create_app()
276
277			with app.app_context():
278			try:
279			process_email_balances()
280			print("Email balance import completed successfully")
281			except Exception as e:
282			print(f"Email balance import failed: {e}")
283			sys.exit(1)
284

whahn1983 / pycashflow

Push — master ( cd6825...32c582 )

app.getemail.send_new_user_notification() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like