我将为您创建一个完整的社区二手图书交换小程序系统。这个系统基于创新创业理论，旨在解决图书资源闲置和浪费问题。
项目结构
community_book_exchange/
├── main.py # 主程序入口
├── user_manager.py # 用户管理模块
├── book_manager.py # 图书管理模块
├── matching_engine.py # 智能匹配引擎
├── exchange_manager.py # 交换管理模块
├── map_service.py # 地图服务模块
├── notification_service.py # 通知服务模块
├── database.py # 数据库管理
├── config.py # 配置文件
├── requirements.txt # 依赖包列表
├── README.md # 项目说明
├── static/ # 静态资源
│ ├── css/
│ ├── js/
│ └── images/
└── templates/ # 页面模板
├── index.html
├── login.html
└── exchange.html
1. 主程序 (main.py)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
社区二手图书交换小程序 - 主程序
基于创新创业理论，解决图书闲置浪费问题
功能：图书信息管理、智能匹配、线下交换组织
作者：AI Assistant
日期：2024年
"""
import sys
import os
from datetime import datetime
from user_manager import UserManager
from book_manager import BookManager
from matching_engine import MatchingEngine
from exchange_manager import ExchangeManager
from notification_service import NotificationService
from map_service import MapService
from database import DatabaseManager
class CommunityBookExchangeApp:
def __init__(self):
"""初始化应用程序"""
self.db = DatabaseManager()
self.user_manager = UserManager(self.db)
self.book_manager = BookManager(self.db)
self.matching_engine = MatchingEngine(self.db)
self.exchange_manager = ExchangeManager(self.db)
self.notification_service = NotificationService()
self.map_service = MapService()
self.current_user = None
def show_main_menu(self):
"""显示主菜单"""
print("\n" + "="*50)
print("📚 社区二手图书交换小程序 📚")
print("="*50)
if self.current_user:
print(f"欢迎，{self.current_user['username']}！")
print("1. 用户注册/登录")
print("2. 添加图书信息")
print("3. 查看我的图书")
print("4. 搜索可交换图书")
print("5. 发起交换请求")
print("6. 查看交换记录")
print("7. 社区图书推荐")
print("8. 交换活动管理")
print("9. 退出程序")
print("-"*50)
def handle_user_registration(self):
"""处理用户注册/登录"""
print("\n👥 用户注册/登录")
print("1. 新用户注册")
print("2. 用户登录")
print("3. 返回主菜单")
choice = input("请选择操作：")
if choice == "1":
self.register_new_user()
elif choice == "2":
self.user_login()
elif choice == "3":
return
else:
print("❌ 无效选择")
def register_new_user(self):
"""注册新用户"""
print("\n📝 新用户注册")
username = input("请输入用户名：")
phone = input("请输入手机号：")
address = input("请输入详细地址：")
community = input("请输入所在小区名称：")
# 验证用户信息
if not all([username, phone, address, community]):
print("❌ 所有字段都必须填写")
return
# 注册用户
user_id = self.user_manager.register_user({
'username': username,
'phone': phone,
'address': address,
'community': community
})
if user_id:
print(f"✅ 注册成功！用户ID：{user_id}")
# 自动登录
self.current_user = self.user_manager.login_by_phone(phone)
else:
print("❌ 注册失败，用户名可能已存在")
def user_login(self):
"""用户登录"""
print("\n🔑 用户登录")
phone = input("请输入手机号：")
user = self.user_manager.login_by_phone(phone)
if user:
self.current_user = user
print(f"✅ 登录成功！欢迎回来，{user['username']}")
else:
print("❌ 登录失败，用户不存在")
def add_book(self):
"""添加图书信息"""
if not self.current_user:
print("❌ 请先登录")
return
print("\n📖 添加图书信息")
title = input("书名：")
author = input("作者：")
isbn = input("ISBN（可选）：")
category = input("图书类别：")
condition = input("图书状况（全新/良好/一般/较差）：")
description = input("图书描述：")
book_data = {
'user_id': self.current_user['id'],
'title': title,
'author': author,
'isbn': isbn,
'category': category,
'condition': condition,
'description': description,
'status': 'available'
}
book_id = self.book_manager.add_book(book_data)
if book_id:
print(f"✅ 图书添加成功！图书ID：{book_id}")
# 自动寻找匹配
matches = self.matching_engine.find_matches(book_id)
if matches:
print(f"🎉 找到 {len(matches)} 个潜在交换对象！")
self.show_matches(matches)
else:
print("❌ 图书添加失败")
def show_my_books(self):
"""显示我的图书"""
if not self.current_user:
print("❌ 请先登录")
return
books = self.book_manager.get_user_books(self.current_user['id'])
if not books:
print("📭 您还没有添加任何图书")
return
print(f"\n📚 您的图书列表（共{len(books)}本）：")
print("-" * 80)
for i, book in enumerate(books, 1):
status_icon = "✅" if book['status'] == 'available' else "🔄"
print(f"{i}. {status_icon} 《{book['title']}》- {book['author']}")
print(f" 类别：{book['category']} | 状况：{book['condition']}")
print(f" 添加时间：{book['created_at']}")
print()
def search_books(self):
"""搜索可交换图书"""
if not self.current_user:
print("❌ 请先登录")
return
print("\n🔍 搜索可交换图书")
keyword = input("请输入搜索关键词（书名/作者）：")
if not keyword:
print("❌ 请输入搜索关键词")
return
books = self.book_manager.search_available_books(keyword)
if not books:
print("📭 没有找到匹配的图书")
return
print(f"\n📖 搜索结果（共{len(books)}本）：")
print("-" * 80)
for i, book in enumerate(books, 1):
owner_name = self.user_manager.get_user_name(book['user_id'])
distance = self.map_service.calculate_distance(
self.current_user['address'],
book['owner_address']
)
print(f"{i}. 《{book['title']}》- {book['author']}")
print(f" 所有者：{owner_name} | 距离：{distance:.1f}km")
print(f" 类别：{book['category']} | 状况：{book['condition']}")
print(f" 描述：{book['description'][:50]}...")
print()
def show_matches(self, matches):
"""显示匹配结果"""
print("\n🎯 智能匹配结果：")
print("-" * 60)
for i, match in enumerate(matches[:5], 1): # 只显示前5个最佳匹配
target_book = self.book_manager.get_book(match['target_book_id'])
target_user = self.user_manager.get_user(match['target_user_id'])
compatibility = match['compatibility_score']
print(f"{i}. 《{target_book['title']}》- {target_book['author']}")
print(f" 交换兼容性：{compatibility:.1%}")
print(f" 所有者：{target_user['username']}")
print(f" 距离：{match['distance']:.1f}km")
print()
def initiate_exchange(self):
"""发起交换请求"""
if not self.current_user:
print("❌ 请先登录")
return
print("\n💌 发起交换请求")
my_book_id = input("请输入您要交换的图书ID：")
target_book_id = input("请输入目标图书ID：")
try:
my_book_id = int(my_book_id)
target_book_id = int(target_book_id)
except ValueError:
print("❌ 图书ID必须是数字")
return
# 创建交换请求
exchange_id = self.exchange_manager.create_exchange_request(
self.current_user['id'], my_book_id, target_book_id
)
if exchange_id:
print(f"✅ 交换请求已发送！交换ID：{exchange_id}")
# 发送通知
target_user_id = self.book_manager.get_book_owner(target_book_id)
self.notification_service.send_exchange_notification(
target_user_id, exchange_id
)
else:
print("❌ 交换请求发送失败")
def show_exchange_history(self):
"""显示交换历史"""
if not self.current_user:
print("❌ 请先登录")
return
exchanges = self.exchange_manager.get_user_exchanges(self.current_user['id'])
if not exchanges:
print("📭 您还没有任何交换记录")
return
print(f"\n📋 您的交换记录（共{len(exchanges)}条）：")
print("-" * 80)
for exchange in exchanges:
status_icon = {
'pending': '⏳',
'accepted': '✅',
'rejected': '❌',
'completed': '🎉'
}.get(exchange['status'], '❓')
my_book = self.book_manager.get_book(exchange['my_book_id'])
target_book = self.book_manager.get_book(exchange['target_book_id'])
print(f"{status_icon} 交换ID：{exchange['id']}")
print(f" 我的图书：《{my_book['title']}》")
print(f" 目标图书：《{target_book['title']}》")
print(f" 状态：{exchange['status']} | 创建时间：{exchange['created_at']}")
print()
def show_recommendations(self):
"""显示社区推荐"""
if not self.current_user:
print("❌ 请先登录")
return
recommendations = self.matching_engine.get_community_recommendations(
self.current_user['id']
)
if not recommendations:
print("📭 暂无推荐图书")
return
print(f"\n🌟 为您推荐的图书（共{len(recommendations)}本）：")
print("-" * 80)
for i, rec in enumerate(recommendations[:10], 1):
book = self.book_manager.get_book(rec['book_id'])
user = self.user_manager.get_user(book['user_id'])
print(f"{i}. 《{book['title']}》- {book['author']}")
print(f" 推荐理由：{rec['reason']}")
print(f" 所有者：{user['username']} | 相似度：{rec['similarity']:.1%}")
print()
def manage_exchange_events(self):
"""管理交换活动"""
print("\n🎪 交换活动管理")
print("1. 查看附近交换活动")
print("2. 创建交换活动")
print("3. 返回主菜单")
choice = input("请选择操作：")
if choice == "1":
self.show_nearby_events()
elif choice == "2":
self.create_exchange_event()
elif choice == "3":
return
else:
print("❌ 无效选择")
def show_nearby_events(self):
"""显示附近交换活动"""
events = self.exchange_manager.get_nearby_events(self.current_user['community'])
if not events:
print("📭 附近暂无交换活动")
return
print(f"\n🎪 附近交换活动（共{len(events)}个）：")
print("-" * 60)
for event in events:
print(f"📅 {event['title']}")
print(f" 时间：{event['event_date']} {event['event_time']}")
print(f" 地点：{event['location']}")
print(f" 参与人数：{event['participant_count']}/{event['max_participants']}")
print()
def create_exchange_event(self):
"""创建交换活动"""
print("\n➕ 创建交换活动")
title = input("活动标题：")
description = input("活动描述：")
location = input("活动地点：")
date = input("活动日期（YYYY-MM-DD）：")
time = input("活动时间（HH:MM）：")
max_participants = input("最大参与人数：")
try:
max_participants = int(max_participants)
except ValueError:
print("❌ 参与人数必须是数字")
return
event_data = {
'organizer_id': self.current_user['id'],
'title': title,
'description': description,
'location': location,
'event_date': date,
'event_time': time,
'max_participants': max_participants,
'community': self.current_user['community']
}
event_id = self.exchange_manager.create_event(event_data)
if event_id:
print(f"✅ 活动创建成功！活动ID：{event_id}")
else:
print("❌ 活动创建失败")
def run(self):
"""运行主程序"""
print("🚀 启动社区二手图书交换小程序...")
while True:
try:
self.show_main_menu()
choice = input("请选择操作（1-9）：")
if choice == "1":
self.handle_user_registration()
elif choice == "2":
self.add_book()
elif choice == "3":
self.show_my_books()
elif choice == "4":
self.search_books()
elif choice == "5":
self.initiate_exchange()
elif choice == "6":
self.show_exchange_history()
elif choice == "7":
self.show_recommendations()
elif choice == "8":
self.manage_exchange_events()
elif choice == "9":
print("👋 感谢使用，再见！")
break
else:
print("❌ 无效选择，请重新输入")
input("\n按回车键继续...")
except KeyboardInterrupt:
print("\n\n👋 程序已退出")
break
except Exception as e:
print(f"❌ 发生错误：{e}")
if __name__ == "__main__":
app = CommunityBookExchangeApp()
app.run()
2. 用户管理模块 (user_manager.py)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
用户管理模块
负责用户注册、登录、信息管理等功能
基于创新创业理论中的用户中心设计理念
"""
import hashlib
import uuid
from datetime import datetime
class UserManager:
def __init__(self, db):
self.db = db
self.current_user = None
def register_user(self, user_data):
"""注册新用户"""
# 验证数据完整性
required_fields = ['username', 'phone', 'address', 'community']
for field in required_fields:
if not user_data.get(field):
raise ValueError(f"缺少必要字段：{field}")
# 检查手机号是否已注册
existing_user = self.db.execute_query(
"SELECT id FROM users WHERE phone = ?",
(user_data['phone'],)
)
if existing_user:
return None # 手机号已存在
# 检查用户名是否已存在
existing_username = self.db.execute_query(
"SELECT id FROM users WHERE username = ?",
(user_data['username'],)
)
if existing_username:
return None # 用户名已存在
# 生成用户ID和密码哈希
user_id = str(uuid.uuid4())
password_hash = self._hash_password(user_data['phone']) # 使用手机号作为初始密码
# 插入用户数据
query = """
INSERT INTO users (id, username, phone, password_hash, address, community, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
"""
params = (
user_id,
user_data['username'],
user_data['phone'],
password_hash,
user_data['address'],
user_data['community'],
datetime.now()
)
success = self.db.execute_insert(query, params)
return user_id if success else None
def login_by_phone(self, phone):
"""通过手机号登录"""
query = """
SELECT id, username, phone, address, community, created_at
FROM users WHERE phone = ?
"""
result = self.db.execute_query(query, (phone,))
if result:
self.current_user = dict(result[0])
return self.current_user
return None
def login_by_username(self, username, password):
"""通过用户名和密码登录"""
password_hash = self._hash_password(password)
query = """
SELECT id, username, phone, address, community, created_at
FROM users WHERE username = ? AND password_hash = ?
"""
result = self.db.execute_query(query, (username, password_hash))
if result:
self.current_user = dict(result[0])
return self.current_user
return None
def update_user_info(self, user_id, update_data):
"""更新用户信息"""
allowed_fields = ['username', 'address', 'community']
updates = []
params = []
for field, value in update_data.items():
if field in allowed_fields and value:
updates.append(f"{field} = ?")
params.append(value)
if not updates:
return False
params.append(user_id)
query = f"UPDATE users SET {', '.join(updates)} WHERE id = ?"
return self.db.execute_update(query, params)
def get_user(self, user_id):
"""获取用户信息"""
query = """
SELECT id, username, phone, address, community, created_at
FROM users WHERE id = ?
"""
result = self.db.execute_query(query, (user_id,))
return dict(result[0]) if result else None
def get_user_name(self, user_id):
"""获取用户姓名"""
user = self.get_user(user_id)
return user['username'] if user else "未知用户"
def get_users_by_community(self, community):
"""获取同小区用户"""
query = """
SELECT id, username, phone, address
FROM users WHERE community = ?
"""
results = self.db.execute_query(query, (community,))
return [dict(row) for row in results]
def delete_user(self, user_id):
"""删除用户（软删除）"""
query = "UPDATE users SET is_active = 0 WHERE id = ?"
return self.db.execute_update(query, (user_id,))
def _hash_password(self, password):
"""密码哈希加密"""
return hashlib.sha256(password.encode()).hexdigest()
def validate_user_data(self, user_data):
"""验证用户数据有效性"""
errors = []
# 验证手机号格式
if not self._validate_phone(user_data.get('phone', '')):
errors.append("手机号格式不正确")
# 验证用户名长度
username = user_data.get('username', '')
if len(username) < 2 or len(username) > 20:
errors.append("用户名长度应在2-20个字符之间")
# 验证必填字段
required_fields = ['username', 'phone', 'address', 'community']
for field in required_fields:
if not user_data.get(field):
errors.append(f"缺少必要字段：{field}")
return errors
def _validate_phone(self, phone):
"""验证手机号格式"""
import re
pattern = r'^1[3-9]\d{9}$'
return re.match(pattern, phone) is not None
3. 图书管理模块 (book_manager.py)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
图书管理模块
负责图书信息的增删改查、状态管理等
体现共享经济理念，促进资源循环利用
"""
from datetime import datetime
class BookManager:
def __init__(self, db):
self.db = db
def add_book(self, book_data):
"""添加图书信息"""
# 验证必要字段
required_fields = ['user_id', 'title', 'author', 'category']
for field in required_fields:
if not book_data.get(field):
raise ValueError(f"缺少必要字段：{field}")
# 生成图书ID
book_id = str(uuid.uuid4())
# 插入图书数据
query = """
INSERT INTO books (id, user_id, title, author, isbn, category, condition,
description, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
params = (
book_id,
book_data['user_id'],
book_data['title'],
book_data['author'],
book_data.get('isbn', ''),
book_data['category'],
book_data.get('condition', '良好'),
book_data.get('description', ''),
book_data.get('status', 'available'),
datetime.now(),
datetime.n
关注我，有更多实用程序等等你！
