AI CRM 툴 개발 기록 (1) 흩어진 고객 데이터를 모으게 된 이유와 전체 구조

엑셀 세 개를 합치던 마케터의 하루 MotherK에서 "지난달에 두 번 이상 산 고객에게 신상품 알림톡을 보내자"는 이야기가 나오면, 실제로 그 명단을 만드는 데 반나절이 걸렸다. 고객 데이터가 한곳에 있지 않았기 때문이다. 자사몰은 카페24에, 스마트스토어 주문은 네이버에, 멘토리 회원과 주문은 별도 MySQL DB에 있었다. 담당자는 채널마다 엑셀을 내려받고, 전화번호 형식을 맞추고, 중복을 눈으로 지운 다음 발송 화면에 붙여 넣었다. 이 과정에는 반복 작업 말고도 문제가 있었다. 같은 조건으로 명단을 다시 뽑으면 매번 결과가 조금씩 달랐고, 조건 자체가 담당자의 머릿속에만 있었다. "충성 고객"의 기준이 3회 구매인지 50만 원 이상인지는 그때그때 달랐다. 그래서 고객·주문 데이터를 매일 자동으로 한곳에 모으고, 조건을 저장해 두고 다시 쓸 수 있는 사내 CRM 콘솔을 직접 만들기로 했다. 이 글은 시리즈의 첫 편으로, 왜 만들었는지와 전체 구조, 그리고 처음 뼈대를 세우면서 부딪힌 문제를 다룬다. 네이버 주문의 개인정보 마스킹 문제와 SQLite에서 PostgreSQL로 옮긴 과정은 다음 편에서 따로 정리했다. 해야 했던 일과 걸림돌 요구사항은 단순하게 정리됐다. 카페24 자사몰, 네이버 스마트스토어, 멘토리 세 곳의 고객·주문을 매일 자동으로 가져온다. 세그먼트(VIP/충성/일반/신규)와 활동 상태(활성/위험/이탈)를 자동으로 분류한다. 조건을 조합해 "오디언스"를 만들고 저장하며, 대상자 수가 어떻게 변하는지 추적한다. 만든 오디언스로 카카오 알림톡·브랜드 메시지를 바로 보낸다. 어려운 점은 데이터를 가져오는 방법이 채널마다 달랐다는 것이다. 카페24는 공식 API가 있었고, 네이버는 커머스 API가 있었지만 서명 방식과 조회 범위 제한이 까다로웠다. 멘토리는 API 없이 DB에 직접 붙어야 했다. 유입·퍼널 같은 통계는 API가 아예 없어서 관리 화면에서 CSV를 내려받는 수밖에 없었다. 또 사용자는 개발자가 아니라 마케터였기 때문에, 필터 조건을 JSON으로 쓰게 할 수는 없었다. 구조: Flask 한 덩어리 + 새벽 배치 사내 몇 명이 쓰는 도구라 마이크로서비스로 쪼갤 이유가 없었다. Flask 3.1 앱 하나에 라우트를 모두 두고, SQLAlchemy로 모델을 정의했다. 무거운 수집 작업은 웹 요청 안에서 돌리지 않고 APScheduler가 새벽에 실행하도록 했다. 크롤러처럼 오래 걸리고 메모리를 많이 쓰는 스크립트는 subprocess 로 별도 프로세스를 띄웠다. 크롤러가 죽어도 웹 서버는 영향을 받지 않게 하기 위해서다. 데이터 출처 가져오는 방법 실행 시점 카페24 자사몰 공식 API + 통계 화면 CSV(Playwright) 새벽 1시대 멘토리 MySQL 직접 조회(PyMySQL) 후 집계 테이블 적재 새벽 1시 40분대 네이버 스마트스토어 커머스 API, 최근 3일 재동기화 오전 6시 파생 데이터 최종 주문일 재계산, 자동 오디언스 캐시 갱신 새벽 2시대 작업 순서에도 의미가 있다. 원천 데이터를 먼저 적재하고, 누락된 날짜를 메우는 백필 작업을 2시에 돌리고, 그다음에 최종 주문일과 자동 오디언스 캐시를 다시 계산한다. 시간은 코드에 고정하지 않고 CRON_<JOB_ID>=HH:MM 환경변수로 덮어쓸 수 있게 했다. 외부 서비스 점검 시간과 겹칠 때 배포 없이 시간을 옮기기 위해서였다. 사용자가 버튼을 눌러 시작하는 수동 동기화는 별도의 작업 관리자가 맡는다. 작업마다 ID를 발급하고 진행 상황을 JSON 파일에 기록해서, 페이지를 옮겨 다녀도 상단 위젯에 "주문 수집 중... 15/90일" 같은 진행률이 계속 보이도록 했다. 시행착오 1. 새벽 크롤링이 여러 번 돌았다 스케줄러는 앱이 시작될 때 BackgroundScheduler 를 만들어 띄우는 방식이었다. 개발 서버에서는 프로세스가 하나라 문제가 없었다. 하지만 운영에서는 Gunicorn이 워커를 여러 개 띄우고, 워커마다 app.py 를 import한다. 워커 수만큼 스케줄러가 생기고, 같은 시각에 같은 크롤링 스크립트가 여러 번 실행되는 구조였다. 같은 관리 화면에 동시에 로그인해 CSV를 받고, 같은 날짜 데이터를 동시에 적재하는 상황은 막아야 했다. 워커 사이에 메모리를 공유하지 않으니, 파일 시스템을 락으로 썼다. 임시 디렉터리에 락 파일을 "x" 모드(이미 있으면 실패)로 만들고 PID를 적는다. 파일이 이미 있으면 그 PID의 프로세스가 살아 있는지 os.kill(pid, 0) 으로 확인하고, 죽은 프로세스의 락이면 지우고 다시 시도한다. 락을 잡은 워커 하나만 스케줄러를 띄운다. _lock_path = os.path.join(tempfile.gettempdir(), "crm_scheduler.lock") _lock_acquired = False try: if os.path.exists(_lock_path): with open(_lock_path) as _lf: _existing_pid = int(_lf.read().strip()) try: os.kill(_existing_pid, 0) # 살아 있는지만 확인 _stale = False except (OSError, SystemError): _stale = True if _stale: os.unlink(_lock_path) # 죽은 워커의 락은 정리 with open(_lock_path, "x") as _lf: # 이미 있으면 FileExistsError _lf.write(str(os.getpid())) _lock_acquired = True except (FileExistsError, ValueError, OSError): pass if _lock_acquired: _scheduler = BackgroundScheduler() ... 같은 문제가 초기 관리자 계정 생성에서도 나왔다. 여러 워커가 동시에 "관리자 계정이 없으면 만든다"를 실행하면서 같은 사용자명을 두 번 넣으려 했다. 여기서는 락 대신, 커밋이 실패하면 롤백하고 계정이 실제로 생겼는지 다시 확인해 있으면 그냥 넘어가도록 했다. 한 번만 실행돼야 하는 코드는 "워커가 여러 개"라는 전제에서 다시 봐야 한다는 걸 이때 배웠다. 2. 마케터가 JSON을 쓸 수는 없다: 자연어 → 필터 오디언스 화면에는 세그먼트 칩, 주문 횟수·금액 범위, 마지막 주문 후 경과일, 주문 기간, 전화번호 유무 같은 필터가 있다. 조건이 많아지자 원하는 고객군을 만들려면 필터 여러 개를 조합해야 했고, 쓰는 사람마다 조합이 달랐다. 그래서 "어제 주문한 2회 이상 구매 고객"처럼 문장으로 쓰면 필터를 채워 주는 기능을 GPT-4o-mini로 붙였다. 먼저 해결해야 할 것은 날짜였다. 모델은 오늘이 며칠인지 모르기 때문에 "어제", "이번 달" 같은 상대 표현을 정확한 날짜로 바꿀 수 없다. 그래서 시스템 프롬프트에 오늘 날짜를 직접 넣고, 쓸 수 있는 필터 키와 각 세그먼트의 기준(예: VIP는 5회 이상 또는 50만 원 이상)을 함께 적었다. "어제 구매한 고객"이면 ordered_after 와 ordered_before 에 같은 날짜를 넣으라는 예시도 프롬프트에 두었다. 두 번째는 응답 형식이었다. "JSON만 응답하라"고 지시해도 모델이 마크다운 코드 블록으로 감싸서 돌려주는 경우가 있고, 그러면 json.loads 가 실패한다. temperature를 0.3으로 낮추고, 응답이 백틱으로 시작하면 첫 줄과 마지막 펜스를 잘라 내도록 했다. content = result["choices"][0]["message"]["content"].strip() if content.startswith("```"): content = content.split("\n", 1)[1].rsplit("```", 1)[0] parsed = json.loads(content) filters = parsed.get("filters", {}) q = apply_naver_audience_filter(filters) # 사람이 쓰는 필터와 같은 함수 parsed["count"] = q.count() parsed["phone_count"] = q.filter( NaverCustomer.phone.isnot(None), NaverCustomer.phone != "", ~NaverCustomer.phone.like("%*%")).count() return jsonify(parsed) 가장 신경 쓴 부분은 AI 결과를 바로 저장하지 않는 것이었다. 모델이 만든 JSON은 사람이 필터를 고를 때와 같은 apply_*_audience_filter 함수를 통과하고, 화면에는 대상자 수와 실제로 문자를 보낼 수 있는 전화번호 확보 수가 함께 나온다. 사용자가 "필터에 적용"을 누르면 결과가 필터 UI에 채워지고, 거기서 칩을 빼거나 범위를 고친 뒤 저장한다. 모델이 모르는 키를 만들어도 필터 함수가 무시하기 때문에 쿼리가 깨지지 않는다. 프롬프트로 SQL을 만들게 하는 방식은 처음부터 고려하지 않았다. 필터 키를 정해 두고 그 안에서만 고르게 하는 편이 검증도 쉽고 사고도 적다. 3. 알림톡 이미지 업로드가 404를 반환했다 카카오 알림톡 템플릿 등록 화면을 만들 때, 이미지형 템플릿의 이미지를 NCP SENS API로 올리려 했다. 그런데 업로드 요청에 404가 돌아왔다. 발송 기능을 업로드 API에 묶어 둘 수는 없어서, 업로드한 이미지를 서버의 정적 디렉터리( /static/uploads/kakao/ )에 저장하고, 템플릿에는 그 URL을 넣는 방식으로 바꿨다. 같은 작업에서 템플릿 목록 조회 파라미터도 page/size 가 아니라 문서 기준인 pageIndex/pageSize 였다는 것을 찾아 고쳤다. 외부 API는 문서 예시를 그대로 믿기보다 한 번씩 직접 호출해 확인해야 한다는 걸 이때 다시 느꼈다. 결과 세 채널의 고객과 주문이 매일 새벽 한곳에 쌓이고, 마케터는 오디언스 화면에서 조건을 고르거나 문장으로 설명해 대상자를 만든다. 저장된 오디언스는 페이지를 열 때마다 대상자 수를 다시 계산하고, 이전 대비 증감 배지(+12 / -5 같은 형태)와 날짜별 변동 이력을 보여 준다. 만든 오디언스는 알림톡이나 브랜드 메시지 발송 화면에서 바로 고를 수 있어서, 엑셀을 내려받아 합치는 과정이 없어졌다. 회고 app.py 하나에 모든 라우트를 넣은 것 은 초기 속도에는 도움이 됐지만, 채널이 늘면서 파일이 너무 커졌다. 카페24·네이버·멘토리마다 비슷한 오디언스 필터와 AI 추천 함수가 따로 생겼다. 다시 만든다면 처음부터 채널별 Blueprint로 나누고, 필터 정의를 공통 스키마로 묶어 AI 프롬프트도 그 스키마에서 자동으로 만들겠다. 파일 락 기반 스케줄러 는 서버 한 대에서는 충분했지만, 서버가 늘어나면 통하지 않는다. 그때는 스케줄러를 웹 프로세스에서 떼어 별도 프로세스로 돌리거나 DB 락을 쓰는 편이 맞다. AI는 필터를 채우는 역할까지만 맡긴 것은 잘한 결정이었다. 결과가 이상해도 사람이 화면에서 바로 보고 고칠 수 있었다. 기간: 2026.04 ~ 2026.09 기술: Python, Flask 3.1, SQLAlchemy, PostgreSQL, APScheduler, Gunicorn, Playwright, PyMySQL, OpenAI GPT-4o-mini, 카페24 API, 네이버 커머스 API, NCP SENS 관련 프로젝트: AI CRM 툴