Python Flask构建客户关系管理系统:数据模型到上线实践
2026/9/14 13:53:29 网站建设 项目流程

简介:一份基于Python的客户关系管理系统(CRM)项目资源,面向具备Python基础、希望掌握Web开发或企业级应用开发的读者。项目围绕客户信息管理、销售流程自动化、客户服务与数据分析等核心模块展开,覆盖从数据库设计到前端页面呈现的完整链路,可作为课程设计、毕业设计或企业内训的参考范本。压缩包共247个文件,大小仅4.35MB,核心为62个Python源码文件,并配套28个SCSS/LESS样式、21个HTML页面、19个CSS及12个JS等前端资源,另含一个SQLite数据库文件,便于直接运行与二次开发。资源还体现了用户权限控制、数据库交互、API接口扩展等实战要点,有助于理解CRM系统的完整架构。目前已有221人学习下载,适合需要快速理解系统结构或在此基础上扩展功能的开发者。通过该资源可以学习如何利用Python生态实现客户数据的增删改查、销售进度跟踪与可视化报表,掌握Django/Flask等框架在真实项目中的组织方式,同时借鉴前端样式与交互设计,提升全栈开发与项目落地能力。

1. 一张Excel放不下客户跟进记录时,就是Python客户关系管理系统要接管的时候

客户名单在销售手里是一份Excel,在老板手里是另一份Excel,在售后那边还单独维护着一批。同一个客户被录两遍,跟进记录要么写进备注栏,要么散落在聊天记录里。用Python写一套客户关系管理系统,不是要去对标Salesforce,而是解决“客户数据只有一个入口、改动可追溯、跟进提醒不靠人脑”这三件事。它的优势不在性能,在快:Flask加几十行SQL就能跑起第一个可用版本,后面再逐步补权限、去重和统计。按我落地内部系统的顺序来:先校验数据模型,客户表、联系人表和跟进记录表一开始就分开;再写登录、列表、搜索的最小闭环;然后把权限越权和慢查询单独拿出来处理;最后补上线前必须有的定时提醒、数据库迁移和备份。

2. 客户关系管理系统数据模型:客户表、联系人表与跟进记录表的职责划分

数据模型错了,后面每写一个查询都在为错误决定付利息。很多半路转Python的团队做CRM,第一版就把所有字段塞进一张客户表:联系人写一个varchar,跟进备注再写一个text,看起来表少、简单,等客户超过一千、一条客户记录底下的跟进历史有几十条时,UPDATE频繁撞锁,查询越来越慢。先花一晚上把表拆对,后面能省一周调试时间。

2.1 客户表与联系人表:为什么必须拆成两张表

客户是公司或组织,联系人是在这个公司里和你沟通的具体的人。一个客户通常对应三五个联系人。联系人字段如果放进客户表,要么把几个人名拼进一个varchar,要么一条客户记录重复存三行,name和phone数据互相矛盾。我在最小系统里把两张表拆开:客户表存组织级属性,联系人表通过customer_id指向客户。

客户主表字段如下:

CREATE TABLE customer ( id BIGSERIAL PRIMARY KEY, name VARCHAR(255) NOT NULL, phone VARCHAR(32), industry VARCHAR(64), source VARCHAR(32), owner_user_id INTEGER, status SMALLINT DEFAULT 1, level SMALLINT DEFAULT 0, created_at TIMESTAMP NOT NULL DEFAULT NOW(), updated_at TIMESTAMP NOT NULL DEFAULT NOW() );

字段选择是有意的。status表示当前商机阶段,1-潜在、2-跟进中、3-成交、4-流失,用SMALLINT存编码而不是存汉字,是为了以后改状态显示名时不动数据库。level是ABC客户等级,用来给报表的加权成交额计算提供依据。owner_user_id是数据归属,第三章的权限过滤就靠它。

联系人表补一个独立主键和基础字段:

CREATE TABLE contact ( id BIGSERIAL PRIMARY KEY, customer_id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, phone VARCHAR(32), position VARCHAR(64), created_at TIMESTAMP NOT NULL DEFAULT NOW() ); CREATE INDEX idx_contact_customer ON contact(customer_id);

在Python端查询某个客户的全部联系人时,一条SELECT * FROM contact WHERE customer_id = %s就能拿到,完全不需要做字符串拆分。

2.2 跟进记录表:时间序列数据用追加写,不要用覆盖写

跟进记录在CRM里是最容易设计错的表。常见做法是客户表上直接放一个follow_content字段,每次跟进就UPDATE一次,旧记录直接被覆盖,事后想复盘客户经历只能靠聊天记录回忆。正确设计是拆一张follow_up表,每次电话、拜访、微信沟通都INSERT一条记录,形成按时间排列的事件流。

CREATE TABLE follow_up ( id BIGSERIAL PRIMARY KEY, customer_id INTEGER NOT NULL, user_id INTEGER NOT NULL, content TEXT NOT NULL, next_time TIMESTAMP, created_at TIMESTAMP NOT NULL DEFAULT NOW() ); CREATE INDEX idx_follow_up_customer ON follow_up(customer_id, created_at DESC);

追加写的好处有两点:一是并发安全,两个销售同时跟进同一个客户时,INSERT不冲突,UPDATE会互相覆盖;二是审计方便,以后想查“这个客户为什么三个月没动”,按customer_id拉出所有记录即可。next_time是给定时提醒任务准备的,每天扫next_time < NOW()的未完成跟进项,答案就在这张表里。

2.3 Flask还是Django:客户关系管理系统框架选型的三个判断点

框架上我默认选Flask,但这不是因为它比Django“轻”这么简单。三个判断点:

  1. 系统体量:客户关系管理系统的主流功能是客户列表、详情、跟进记录和几张统计报表,Django的Admin后台、内置ORM、迁移机制在模型频繁调整时反而成为阻碍。
  2. 谁来看这个代码:内部工具往往只有一两个人维护,Flask整个项目三四个文件,新人翻一遍就能上手;Django的层层约定需要额外学习成本。
  3. 需要什么能力:如果后续要接支付、定时任务、工作流审批,Django的自带生态更省事。

下表是我在这个场景下的倾向。

比较项Flask + 纯SQLDjango + ORM
起步成本一个app.py跑通项目脚手架+模型迁移
改表结构改SQL重启makemigrations+migrate
数据权限自行拼接过滤条件需要配第三方库
适合体量单项目到中小团队全员级或复杂流程
2.3.1 用纯SQL还是ORM,先看团队是否会看执行计划

数据访问层上,我选纯SQL。ORM的价值在对象关系映射,但客户关系管理系统的核心查询几乎都是联表过滤和聚合统计,SQL本身并不复杂,直接用cursor取到字典列表更直观。真正要ORM来兜底的场景是模型特别多、关联关系复杂、团队不会写SQL,这时Django的admin后台能省掉大量开发量。在这个数据模型上加字段时记住一件事:报表字段从源头表取数,不要为了报表快就提前设计一张宽表,等报表真的慢了再考虑物化视图,否则数据一致性会变成每天对账的负担。

3. Flask实现Python客户关系管理系统最小闭环:登录、客户列表与数据权限过滤

模型定完,下一步不是画界面,而是先把最常用的一条链路打通:登录进系统、看到自己名下的客户、能搜索、能打开详情看到跟进记录。这一步走通后,这个系统才叫“能用”,值得继续投入。

3.1 先确认Python环境:版本、pip与虚拟环境

我见过大量系统刚clone下来跑不起来,根因不在代码,在Python环境。先用三条命令确认:

python -V pip -V python -m venv venv && source venv/bin/activate

第一行确认Python版本,代码里如果用到了dict | dict合并语法,3.9以下直接语法报错,这些代码用3.10以上最稳。第二行确认pip能调用,装不上依赖说明PATH里没有Scripts目录,Windows下常见,去系统环境变量里把C:\Python310\Scripts补上即可。第三行建虚拟环境,依赖装进项目目录而不是全局,换机器部署才不会版本互相污染。

提示:Windows下如果pip -V输出乱码或找不到命令,检查系统环境变量里有没有Python的Scripts目录,而不是重装Python。

3.2 后端只做两件事:校验登录态、拼SQL

先接Flask的登录逻辑。密码不存明文,放在sys_user表,pwd_hash字段存werkzeug生成的哈希:

# app.py 最小骨架 from flask import Flask, request, session, redirect, g from werkzeug.security import check_password_hash import sqlite3 app = Flask(__name__) app.secret_key = "change-me-in-production" DATABASE = "crm.db" def get_db(): if "db" not in g: g.db = sqlite3.connect(DATABASE) g.db.row_factory = sqlite3.Row return g.db @app.teardown_appcontext def close_db(exception): db = g.pop("db", None) if db is not None: db.close() @app.route("/login", methods=["POST"]) def login(): username = request.form["username"] password = request.form["password"] user = get_db().execute( "SELECT id, username, pwd_hash, role FROM sys_user WHERE username = ?", (username,), ).fetchone() if user and check_password_hash(user["pwd_hash"], password): session["user_id"] = user["id"] session["role"] = user["role"] return redirect("/customers") return "用户名或密码错误", 401

这个骨架里有三个容易踩的细节。get_db挂在g对象上,请求结束由teardown_appcontext统一关闭,不写close也不会泄漏连接。密码校验用check_password_hash而不是把salt存在另一列再比对,哈希算法本身已经带盐了。session默认由Flask签名的cookie承载,密钥必须换成随机长字符串,否则别人可以伪造会话。

3.3 客户列表:管理员看全部,销售看自己的

最小权限规则是:角色是admin的可以看所有客户,普通销售只能看到owner_user_id等于自己ID的记录。如果业务上还有“客户池”的说法,那就是在查询条件里用OR owner_user_id = 0把未分配客户放出来。下面是列表查询的核心段:

@app.route("/customers") def customers(): keyword = request.args.get("q", "").strip() role = session.get("role") where = [] params = [] if role != "admin": where.append("(c.owner_user_id = ? OR c.owner_user_id = 0)") params.append(session["user_id"]) if keyword: where.append("(c.name LIKE ? OR c.phone LIKE ?)") params.extend([f"%{keyword}%", f"%{keyword}%"]) sql = "SELECT c.*, u.realname AS owner_name FROM customer c LEFT JOIN sys_user u ON u.id = c.owner_user_id" if where: sql += " WHERE " + " AND ".join(where) sql += " ORDER BY c.updated_at DESC LIMIT 20 OFFSET ?" params.append((int(request.args.get("page", 1)) - 1) * 20) rows = get_db().execute(sql, params).fetchall() return render_template("customers.html", customers=rows)

拼接字符串写SQL容易被骂,但这里是动态WHERE条件,所有值都用?占位符传给sqlite3,Python驱动会做转义,SQL注入没有位置可钻。OFFSET参数直接由page算出来,这是最基础的分页方式;等数据量过五万,再换用keyset分页,而不是把OFFSET越翻越大。LEFT JOIN sys_user是为了在列表里显示负责人姓名,不额外查一次用户表。

参数说明建议值
LIMIT每页行数20左右,接口和页面体验都舒服
OFFSET跳过行数(page-1)*20
q搜索关键词姓名或电话统一用LIKE模糊匹配
owner_user_id数据归属0表示公海客户

3.4 客户详情页:一对多关系一次查出来

详情页需要三个数据:客户基本信息、联系人列表、跟进记录列表。这里最常见的问题是为填三项数据发了三次数据库请求。对这个量级的内部工具,三个查询没毛病;但如果哪个接口变得很慢,优先看有没有循环查询——即Python里for循环每个联系人再去查一次跟进记录。在数据量几百到几千时,直接把两次查询的结果在代码里做映射,比ORM里写一堆relationship后懒加载省心。

4. 客户关系管理系统的数据质量与慢查询:越权防护、重复去重与EXPLAIN定位

最小闭环能跑之后,真正消耗时间的往往不是新功能,而是数据本身。权限边界不严、重复客户、SQL没走索引,这三类问题在Excel时代完全不存在,换到管理系统后却每天都有人报。

4.1 越权不只是角色判断:详情页同样要校验归属

上一章的列表页做了角色过滤,但详情页如果只按id查,销售把URL里的id改成别人的客户记录编号就能看到详情。这是客户关系管理系统最常见的安全漏洞。详情和删除接口必须重复执行一次归属校验:

customer = get_db().execute( "SELECT * FROM customer WHERE id = ?", (request.args.get("id"),), ).fetchone() if customer is None: abort(404) if session.get("role") != "admin" and customer["owner_user_id"] not in (session["user_id"], 0): abort(403)

这个判断看起来重复,但必须写,因为列表页的过滤最多只是“不显示”,并不能阻止直接构造URL访问。资源级权限校验应该成为统一装饰器,而不是每个函数复制一遍,否则后面新增几个视图时容易漏。

4.2 重复客户:先识别,再并单

同一个客户被不同销售重复录入,是内部工具上线后必然出现的情况。我一般分两步处理:用phone或name+industry做一致性校验,在写入前拦截;对历史存量数据,用GROUP BY找出疑似重复项再人工确认。

SELECT phone, COUNT(*) FROM customer WHERE phone <> '' GROUP BY phone HAVING COUNT(*) > 1;

这条SQL把同电话的重复记录列出来。并单动作要谨慎,通常不直接DELETE,而是把被合并的客户status改为4,并把其联系人、跟进的customer_id批量UPDATE到保留记录上。两步必须放在同一个数据库事务里,任何一步失败都整体回滚,否则会出现“主客户被保留但跟进记录丢失”的脏状态。

BEGIN; UPDATE follow_up SET customer_id = 1 WHERE customer_id = 2; UPDATE contact SET customer_id = 1 WHERE customer_id = 2; UPDATE customer SET status = 4 WHERE id = 2; COMMIT;

4.3 用EXPLAIN定位慢查询:索引没走对,参数调再大也没用

慢查询不用等用户投诉,先把SQL拉到数据库里执行一遍EXPLAIN就能看。比如搜索“所有姓李的跟进中客户”:

EXPLAIN ANALYZE SELECT * FROM customer WHERE name LIKE '李%' AND status = 2 ORDER BY updated_at DESC LIMIT 20;

如果看到的是Seq Scan,说明索引没覆盖到查询条件。对姓名前缀匹配,普通的B-tree索引有效;如果是LIKE '%李%',前缀带%会让索引失效,需要改成LIKE '李%'或者引入全文检索。客户表现在只有几万行的时候不明显,上百万后慢查询会直接反映在用户体验上。

补一个最常用的复合索引:

CREATE INDEX idx_customer_status_update ON customer(status, updated_at DESC);

这个复合索引用在最常见的列表接口上:WHERE status=2 ORDER BY updated_at DESC可以一次索引返回结果,不需要回表再排序。

症状可能原因解决思路
列表页越翻越慢OFFSET过大改用基于ID的keyset分页
LIKE很慢前缀%导致B-tree失效改为前缀匹配或引入全文检索
UPDATE经常锁并发更新同一记录过多把覆盖写改成追加写
JOIN后排序慢缺少复合索引按WHERE+ORDER字段建联合索引

5. Python客户关系管理系统上线前的三个工程细节:定时提醒、迁移与备份

这套标题对应的系统,生产环境通常跑在Linux服务器上。提醒任务用系统cron比常驻进程省心,数据库从开发期SQLite切到生产PostgreSQL,备份要做在误删之后能回滚的程度。

5.1 用系统cron替代常驻进程的定时跟进提醒

Python写定时任务的库很多,APScheduler是最常见的。但如果是Linux服务器,一条crontab比常驻进程更省心:脚本挂掉就少跑一天,不会出现没人维护的僵尸进程在服务器上反复重启。提醒脚本的设计思路是把“今天到期”和“明天到期”的跟进项一次查出,通过企业微信或钉钉的自定义机器人推给对应负责人:

# reminder.py from datetime import datetime, timedelta import sqlite3, requests conn = sqlite3.connect("crm.db") conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT u.webhook_url, c.name, f.next_time FROM follow_up f JOIN customer c ON c.id = f.customer_id JOIN sys_user u ON u.id = f.user_id WHERE f.next_time BETWEEN ? AND ? """, (datetime.now(), datetime.now() + timedelta(hours=24))).fetchall() for row in rows: requests.post(row["webhook_url"], json={ "msgtype": "text", "text": {"content": f"客户 {row['name']} 需要跟进:{row['next_time']}"} }, timeout=5)

crontab行解释:每天9点10分执行一次提醒任务,日志重定向到reminder.log:

10 9 * * * cd /opt/crm && /opt/crm/venv/bin/python reminder.py >> /var/log/crm_reminder.log 2>&1

这里的要点是先把脚本放在一个能手工执行的目录,直接确认能输出内容再放进cron,否则排错时很难区分是代码问题还是cron环境问题——cron运行时的PATH非常精简,脚本里不要依赖系统环境的PYTHONPATH。

5.2 用增量SQL做迁移,而不是从头导入

如果开发时用SQLite,生产用PostgreSQL,数据库之间的行为差异会成为隐患。不要在“上线”那天用导出导入工具搬数据,而是在项目里维护一个migrations目录,每次变更新建一个带序号的sql文件执行。迁移逻辑简单直接:

psql "$DATABASE_URL" -f migrations/003_add_customer_level.sql

这条命令要求运维人员记住当前执行到哪个文件,对简单项目够用;如果以后要支持回滚,再引入alembic也不迟。

5.3 备份要能回答“误删了怎么办”

备份前先明确:单看备份文件不是目的,能在十分钟内恢复到误删那一刻才是。用PostgreSQL的pg_dump做每日全量备份入库即可;如果操作频繁,需要每半小时做一次增量。恢复时按需还原,不需要把整个数据库导入生产。

pg_dump -Fc crm -f /backup/crm_$(date +%F).dump

定时备份同样交给cron,注意磁盘保留策略:保留最近30天,超过就滚动删除,防止一台服务器上备份文件把磁盘占满。真出问题时,恢复演练至少做一次:在测试库上还原一份dump,确认联系人、跟进记录、客户表都能查询出来,结果为空则备份失败。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询