完成了一个get和一个post的mock请求，抽象通用逻辑，增加返回模板

mock/migrations/0001_initial.py

@@ -0,0 +1,34 @@
+# Generated by Django 4.2.7 on 2023-12-21 01:47
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ZZMockModel',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('ip', models.CharField(help_text='影院ip', max_length=20, verbose_name='影院ip')),
+                ('reportTicket', models.BooleanField(help_text='票房数据上报接口', verbose_name='票房数据上报接口')),
+                ('reportFilmSchedule', models.BooleanField(help_text='排片数据上报接口', verbose_name='排片数据上报接口')),
+                ('uploadScreenSeatInfo', models.BooleanField(help_text='座位信息上报接口', verbose_name='座位信息上报接口')),
+                ('downloadFilmInfo', models.BooleanField(help_text='影片信息下载接口', verbose_name='影片信息下载接口')),
+                ('getCinemaInfo', models.BooleanField(help_text='影院信息下载接口', verbose_name='影院信息下载接口')),
+                ('getScreenInfo', models.BooleanField(help_text='影厅信息下载接口', verbose_name='影厅信息下载接口')),
+                ('getOvertimeTicketStatus', models.BooleanField(help_text='超时票务受理情况查询接口', verbose_name='超时票务受理情况查询接口')),
+                ('validError', models.BooleanField(help_text='数据清洗错误查询接口', verbose_name='数据清洗错误查询接口')),
+            ],
+            options={
+                'verbose_name': '专资接口模拟',
+                'verbose_name_plural': '专资接口模拟',
+                'db_table': 'zz_mock',
+            },
+        ),
+    ]

mock/mock_templates/__init__.py

@@ -0,0 +1,2 @@
+__all__ = ['get_cinema_info_tmp',
+           'report_ticket_tmp']

mock/mock_templates/get_cinema_info_tmp.py

@@ -0,0 +1,45 @@
+import pymysql
+from pymysql.cursors import DictCursor
+
+
+def get(cinema_data):
+    db_config = {
+        'host': cinema_data.ip,
+        'user': cinema_data.db_user,
+        'password': cinema_data.db_pwd,
+        'database': 'cine',
+        'connect_timeout': 5,
+    }
+
+    db_conn = pymysql.Connect(**db_config)
+    db_cursor = db_conn.cursor(cursor=DictCursor)
+    # cinema_info
+    cinema_info_sql_str = 'SELECT * FROM cinema_info;'
+    db_cursor.execute(cinema_info_sql_str)
+    cinema_info_data = db_cursor.fetchone()
+    # cinema_set
+    cinema_set_sql_str = 'SELECT * FROM cinema_set;'
+    db_cursor.execute(cinema_set_sql_str)
+    cinema_set_data = db_cursor.fetchone()
+    # cinema_hall_info
+    cinema_hall_info_sql_str = 'SELECT * FROM cinema_hall_info;'
+    db_cursor.execute(cinema_hall_info_sql_str)
+    cinema_hall_info_data = db_cursor.fetchall()
+    hall_num = len(cinema_hall_info_data)
+    db_conn.close()
+    mock_data = {'message': '操作成功',
+                 'data': {
+                     'cinemaChainName': cinema_info_data['cinema_chain_name'],
+                     'manager': cinema_info_data['cinema_manager'],
+                     'screens': hall_num,
+                     'managerTel': cinema_info_data['cinema_telephone'],
+                     'cinemaName': cinema_set_data['cinema_platform_name'],
+                     'cinemaCode': cinema_set_data['cinema_num'],  # 适配字段
+                     'cinemaLevel': cinema_info_data['cinema_level'],
+                     'businessStatus': '12',  # 定义字段
+                     'officialName': cinema_info_data['official_name'],
+                     'fax': cinema_info_data['cinema_fax']
+                 },
+                 'code': '200',
+                 'status': 'success'}
+    return mock_data

mock/mock_templates/report_ticket_tmp.py

@@ -0,0 +1,3 @@
+def get():
+    mock_data = {'message': '操作成功', 'data': {}, 'code': '200', 'status': 'success'}
+    return mock_data

mock/models.py

@@ -1,3 +1,25 @@
 from django.db import models
 
+
 # Create your models here.
+
+class ZZMockModel(models.Model):
+    ip = models.CharField(verbose_name='影院ip', max_length=20, null=False, help_text='影院ip')
+    reportTicket = models.BooleanField(verbose_name='票房数据上报接口', null=False, help_text='票房数据上报接口')
+    reportFilmSchedule = models.BooleanField(verbose_name='排片数据上报接口', null=False, help_text='排片数据上报接口')
+    uploadScreenSeatInfo = models.BooleanField(verbose_name='座位信息上报接口', null=False,
+                                               help_text='座位信息上报接口')
+    downloadFilmInfo = models.BooleanField(verbose_name='影片信息下载接口', null=False, help_text='影片信息下载接口')
+    getCinemaInfo = models.BooleanField(verbose_name='影院信息下载接口', null=False, help_text='影院信息下载接口')
+    getScreenInfo = models.BooleanField(verbose_name='影厅信息下载接口', null=False, help_text='影厅信息下载接口')
+    getOvertimeTicketStatus = models.BooleanField(verbose_name='超时票务受理情况查询接口', null=False,
+                                                  help_text='超时票务受理情况查询接口')
+    validError = models.BooleanField(verbose_name='数据清洗错误查询接口', null=False, help_text='数据清洗错误查询接口')
+
+    def __str__(self):
+        return self.ip
+
+    class Meta:
+        verbose_name = '专资接口模拟'
+        verbose_name_plural = '专资接口模拟'
+        db_table = 'zz_mock'

mock/urls.py

@@ -19,6 +19,6 @@ from django.urls import path, include
 from mock.views import *
 
 urlpatterns = [
-    # path('report/reportTicket/', report_ticket),
+    path('report/reportTicket', report_ticket),
     path('data/getCinemaInfo', get_cinema_info),
 ]

mock/utils/mock_service.py

@@ -0,0 +1,41 @@
+import requests
+from update.models import Cinema
+from django.http.response import JsonResponse
+from mock.mock_templates import *
+
+ZZ_URL = 'https://zzcs.yinghezhong.com'
+
+
+def mock(_ip, _api, *args, **kwargs):
+    print('mock')
+    cinema_obj = Cinema.objects.filter(ip=_ip).first()
+    if _api == 'getCinemaInfo':
+        mock_data = get_cinema_info_tmp.get(cinema_obj)
+        return JsonResponse(mock_data, json_dumps_params={'ensure_ascii': False})
+    if _api == 'reportTicket':
+        mock_data = report_ticket_tmp.get()
+        return JsonResponse(mock_data, json_dumps_params={'ensure_ascii': False})
+
+def bypass(_r):
+    print('bypass')
+    print(_r.method, _r.path)  # GET /data/getCinemaInfo
+    print(_r.GET)  # <QueryDict: {'cinemaCode': ['14105121'], 'sign': ['f7ca26e79abd3d2225e348da66463fbf'], 'timeStamp': ['1703119043']}>
+    print(_r.POST)
+    if _r.method == 'GET':
+        req_url = ZZ_URL + _r.path
+        req_params = dict(_r.GET)
+        print(req_url)
+        print(req_params)
+        response_dara = requests.get(req_url, params=req_params)
+        print(response_dara.json())
+        return JsonResponse(response_dara.json(), json_dumps_params={'ensure_ascii': False})
+    if _r.method == 'POST':
+        req_url = ZZ_URL + _r.path
+        req_params = dict(_r.GET)
+        req_data = dict(_r.POST)
+        print(req_url)
+        print(req_params)
+        print(req_data)
+        response_dara = requests.post(req_url, params=req_params, data=req_data)
+        print(response_dara.json())
+        return JsonResponse(response_dara.json(), json_dumps_params={'ensure_ascii': False})

mock/views.py

@@ -1,27 +1,44 @@
+import requests
 from django.shortcuts import render
 from django.http.response import JsonResponse
+from mock.utils import mock_service
+from mock.models import ZZMockModel
+from django.views.decorators.csrf import csrf_exempt
 
+ZZ_URL = 'https://zzcs.yinghezhong.com'
 
-# Create your views here.
 
+# 数据上报
 # 1、票房数据上报接口 POST /report/reportTicket
+@csrf_exempt
 def report_ticket(request):
-    if request.method == 'POST':
+    ip = request.META.get('REMOTE_ADDR')
-        print(request.headers)
+    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
-        print(request.body)
+    if mock_service_switch.reportTicket is True:
-        return JsonResponse({'status': 'success'})
+        return mock_service.mock(ip, 'reportTicket')
+    else:
+        return mock_service.bypass(request)
+
+
+# 2、排片数据上报接口 POST /report/reportFilmSchedule
+# 3、座位信息上报接口 POST /report/uploadScreenSeatInfo
 
 
 # 数据下载
 # 1、影片信息下载接口 GET /data/downloadFilmInfo
 # 2、影院信息下载接口 GET /data/getCinemaInfo
 def get_cinema_info(request):
-    print(request.headers)
+    ip = request.META.get('REMOTE_ADDR')
-    print(request.body)
+    mock_service_switch = ZZMockModel.objects.filter(ip=ip).first()
-    if request.method == 'GET':
+    if mock_service_switch.getCinemaInfo is True:
-        print('getCinemaInfo')
+        return mock_service.mock(ip, 'getCinemaInfo')
-        print(request.headers)
+    else:
-        print(request.body)
+        return mock_service.bypass(request)
-        return JsonResponse({'status': 'success'})
-
 # 3、影厅信息下载接口 GET /data/getScreenInfo
+
+
+# 超时票务审批
+# 1、超时票务受理情况查询接口 GET /data/getOvertimeTicketStatus
+#
+# 数据查询
+# 2、数据清洗错误查询接口 POST /query/validError

update/migrations/0003_alter_cinema_options_alter_cinema_client_ver_and_more.py

@@ -0,0 +1,62 @@
+# Generated by Django 4.2.7 on 2023-12-21 01:47
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('update', '0002_rename_num_cinema_inner_id_rename_code_cinema_zz_num_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='cinema',
+            options={'verbose_name': '影院信息', 'verbose_name_plural': '影院信息'},
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='client_ver',
+            field=models.CharField(help_text='鼎新客户端版本', max_length=50, verbose_name='鼎新客户端版本'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='db_pwd',
+            field=models.CharField(help_text='影院数据库密码', max_length=50, verbose_name='影院数据库密码'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='db_user',
+            field=models.CharField(help_text='影院数据库账号', max_length=20, verbose_name='影院数据库账号'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='inner_id',
+            field=models.CharField(help_text='影院4位编码', max_length=4, verbose_name='影院4位编码'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='ip',
+            field=models.CharField(help_text='影院ip', max_length=20, verbose_name='影院ip'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='name',
+            field=models.CharField(help_text='影院名称', max_length=100, verbose_name='影院名称'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='sys_ver',
+            field=models.CharField(help_text='鼎新系统版本', max_length=50, verbose_name='鼎新系统版本'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='update_time',
+            field=models.DateTimeField(auto_now=True, help_text='最后更新时间', verbose_name='最后更新时间'),
+        ),
+        migrations.AlterField(
+            model_name='cinema',
+            name='zz_num',
+            field=models.CharField(help_text='影院转账8位编码', max_length=8, verbose_name='影院转账8位编码'),
+        ),
+    ]