Comprehensive Python Scripts for Property Management System Functions

1. Data Management and Query

Functions for connecting to a MySQL database, executing queries, and performing CRUD operations on the residents table.

import mysql.connector
from tabulate import tabulate

def create_connection(host_name, user_name, user_password, db_name):
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host_name,
            user=user_name,
            passwd=user_password,
            database=db_name
        )
        print("Connection to MySQL DB successful")
    except Error as e:
        print(f"The error '{e}' occurred")
    return connection

def execute_query(connection, query):
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        connection.commit()
        print("Query executed successfully")
    except Error as e:
        print(f"The error '{e}' occurred")

def read_query(connection, query):
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        result = cursor.fetchall()
        return result
    except Error as e:
        print(f"The error '{e}' occurred")

# Connection example
connection = create_connection("localhost", "root", "yourpassword", "property_management")

# CRUD examples

def get_all_residents():
    query = "SELECT * FROM residents"
    results = read_query(connection, query)
    print(tabulate(results, headers=['ID', 'Name', 'Apartment', 'Phone'], tablefmt='grid'))

def add_resident(name, apartment, phone):
    query = f"INSERT INTO residents (name, apartment, phone) VALUES ('{name}', '{apartment}', '{phone}')"
    execute_query(connection, query)

def delete_resident(id):
    query = f"DELETE FROM residents WHERE id = {id}"
    execute_query(connection, query)

def update_resident(id, name, apartment, phone):
    query = f"UPDATE residents SET name = '{name}', apartment = '{apartment}', phone = '{phone}' WHERE id = {id}"
    execute_query(connection, query)

# Sample calls
get_all_residents()
add_resident('张三', 'A栋101', '123456789')
update_resident(1, '李四', 'B栋202', '987654321')
delete_resident(2)

2. Automated Payment Reminders

Script that queries residents with overdue payments and sends email reminders using SMTP.

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import datetime

def send_email(subject, body, to_email):
    from_email = "[email protected]"
    from_password = "your_password"
    msg = MIMEMultipart()
    msg['From'] = from_email
    msg['To'] = to_email
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'plain'))
    server = smtplib.SMTP('smtp.example.com', 587)
    server.starttls()
    server.login(from_email, from_password)
    text = msg.as_string()
    server.sendmail(from_email, to_email, text)
    server.quit()

def get_residents_for_reminder():
    query = "SELECT * FROM residents WHERE last_payment < DATE_SUB(CURDATE(), INTERVAL 1 MONTH)"
    results = read_query(connection, query)
    return results

def send_payment_reminders():
    residents = get_residents_for_reminder()
    for resident in residents:
        name = resident[1]
        email = resident[4]  # assume email is the 5th column
        subject = "缴费提醒"
        body = f"尊敬的{name}，您的物业管理费已到期，请尽快缴纳。"
        send_email(subject, body, email)

# Execute monthly
send_payment_reminders()

3. Monthly Report Generation

Uses pandas to run a SQL query that joins residents and payments, then exports the result to an Excel file.

import pandas as pd
import datetime

def generate_monthly_report():
    query = """
    SELECT 
        r.name, 
        r.apartment, 
        p.amount, 
        p.payment_date 
    FROM 
        residents r 
    JOIN 
        payments p ON r.id = p.resident_id 
    WHERE 
        p.payment_date >= DATE_FORMAT(CURDATE() - INTERVAL 1 MONTH, '%Y-%m-01') AND
        p.payment_date < DATE_FORMAT(CURDATE(), '%Y-%m-01')
    """
    results = read_query(connection, query)
    df = pd.DataFrame(results, columns=['Name', 'Apartment', 'Amount', 'Payment Date'])
    report_path = f"monthly_report_{datetime.datetime.now().strftime('%Y%m')}.xlsx"
    df.to_excel(report_path, index=False)
    print(f"Report generated: {report_path}")

# Generate report
generate_monthly_report()

4. Emergency Notification System

Fetches all resident email addresses and sends a critical alert (e.g., fire drill) via SMTP.

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_emergency_notification(subject, body, to_emails):
    from_email = "[email protected]"
    from_password = "your_password"
    msg = MIMEMultipart()
    msg['From'] = from_email
    msg['To'] = ', '.join(to_emails)
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'plain'))
    server = smtplib.SMTP('smtp.example.com', 587)
    server.starttls()
    server.login(from_email, from_password)
    text = msg.as_string()
    server.sendmail(from_email, to_emails, text)
    server.quit()

def get_all_emails():
    query = "SELECT email FROM residents"
    results = read_query(connection, query)
    emails = [row[0] for row in results if row[0]]
    return emails

def send_emergency_alert():
    subject = "紧急通知：消防演习"
    body = "尊敬的居民，今天下午将进行消防演习，请大家注意安全。"
    emails = get_all_emails()
    send_emergency_notification(subject, body, emails)

# Trigger alert
send_emergency_alert()

5. Maintenance Request Management

Provides functions to add a maintenance request and to list pending requests.

def add_maintenance_request(resident_id, description):
    query = f"INSERT INTO maintenance_requests (resident_id, description, status) VALUES ({resident_id}, '{description}', 'Pending')"
    execute_query(connection, query)

# Example
add_maintenance_request(1, "卫生间漏水")

def get_pending_requests():
    query = "SELECT * FROM maintenance_requests WHERE status = 'Pending'"
    results = read_query(connection, query)
    print(tabulate(results, headers=['ID', 'Resident ID', 'Description', 'Status', 'Created At'], tablefmt='grid'))

# View pending
get_pending_requests()

6. Visitor Management System

Functions to log a visitor entry and to retrieve the most recent visitor records.

def log_visitor(name, contact, purpose, resident_id):
    query = f"INSERT INTO visitors (name, contact, purpose, resident_id, entry_time) VALUES ('{name}', '{contact}', '{purpose}', {resident_id}, NOW())"
    execute_query(connection, query)

# Example
log_visitor('王五', '123456789', '拜访朋友', 1)

def get_recent_visitors():
    query = "SELECT * FROM visitors ORDER BY entry_time DESC LIMIT 10"
    results = read_query(connection, query)
    print(tabulate(results, headers=['ID', 'Name', 'Contact', 'Purpose', 'Resident ID', 'Entry Time'], tablefmt='grid'))

# View recent visitors
get_recent_visitors()

7. Facility Booking System

Allows residents to book facilities and to view existing bookings for a specific facility.

def book_facility(resident_id, facility_id, start_time, end_time):
    query = f"INSERT INTO facility_bookings (resident_id, facility_id, start_time, end_time) VALUES ({resident_id}, {facility_id}, '{start_time}', '{end_time}')"
    execute_query(connection, query)

# Example booking
book_facility(1, 1, '2023-10-15 10:00:00', '2023-10-15 12:00:00')

def get_facility_bookings(facility_id):
    query = f"SELECT * FROM facility_bookings WHERE facility_id = {facility_id} ORDER BY start_time"
    results = read_query(connection, query)
    print(tabulate(results, headers=['ID', 'Resident ID', 'Facility ID', 'Start Time', 'End Time'], tablefmt='grid'))

# View bookings
get_facility_bookings(1)

8. Parking Management System

Records vehicle entry and exit events and provides a function to list logs for a given vehicle.

def log_vehicle_entry_exit(vehicle_id, action, time=None):
    if not time:
        time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    query = f"INSERT INTO vehicle_logs (vehicle_id, action, time) VALUES ({vehicle_id}, '{action}', '{time}')"
    execute_query(connection, query)

# Record examples
log_vehicle_entry_exit(1, 'entry')
log_vehicle_entry_exit(1, 'exit')

def get_vehicle_logs(vehicle_id):
    query = f"SELECT * FROM vehicle_logs WHERE vehicle_id = {vehicle_id} ORDER BY time"
    results = read_query(connection, query)
    print(tabulate(results, headers=['ID', 'Vehicle ID', 'Action', 'Time'], tablefmt='grid'))

# View logs
get_vehicle_logs(1)

9. Announcement Publishing System

Functions to post a new announcement and to retrieve the latest announcements.

def post_announcement(title, content):
    query = f"INSERT INTO announcements (title, content, posted_at) VALUES ('{title}', '{content}', NOW())"
    execute_query(connection, query)

# Example
post_announcement('中秋节活动通知', '尊敬的居民，中秋节将举行庆祝活动，请大家积极参与。')

def get_latest_announcements():
    query = "SELECT * FROM announcements ORDER BY posted_at DESC LIMIT 10"
    results = read_query(connection, query)
    print(tabulate(results, headers=['ID', 'Title', 'Content', 'Posted At'], tablefmt='grid'))

# Retrieve
get_latest_announcements()

10. Database Backup

Simple script that creates a backup directory if needed and runs a mysqldump command to export the property_management database.

import os
import subprocess
import datetime

def backup_database():
    backup_dir = "/path/to/backup/directory"
    if not os.path.exists(backup_dir):
        os.makedirs(backup_dir)
    backup_file = os.path.join(backup_dir, f"db_backup_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.sql")
    command = f"mysqldump -u root -pyourpassword property_management > {backup_file}"
    subprocess.run(command, shell=True)
    print(f"Database backed up to {backup_file}")

# Execute backup
backup_database()

Summary

The collection of scripts above equips property management staff with essential backend functionalities, including resident data CRUD, automated payment reminders, monthly financial reporting, emergency alerts, maintenance request handling, visitor logging, facility reservations, vehicle entry/exit tracking, announcement publishing, and reliable database backup.