DigitalMgnt
/
api-datamanager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update 企业搜索接口

This commit is contained in:
P3ngSaM 2022-08-30 17:04:17 +08:00
parent cb33772174
commit 80c4cbc5da
18 changed files with 586 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
Company/Basic/BasicImpl.py
View File

@ -1,3 +1,4 @@
from Company.CompanyUtils import CompanyUtils
from DBHelper.MongoHelperInstance import DB_TYC, DB_TEST
from Utils.CommonUtil import CommonUtils
@ -9,41 +10,106 @@ class BasicImpl(object):
@staticmethod
def basic_information(**kwargs):
"""工商信息"""
name = kwargs['name']
cid = CompanyUtils.make_new_cid()
condition = kwargs['condition']
record = DB_TYC.find_single_data(
'公司背景',
'基本信息',
{'企业名称': name},
['基本信息']
)
def save_data(**args):
"""查询保存数据"""
company_name = args['cname']
company_id = args['cid']
record = DB_TYC.find_single_data(
'公司背景',
'基本信息',
{'企业名称': company_name},
['基本信息']
)
if record:
record = record['基本信息']
data = dict()
data['企业ID'] = company_id
data['企业名称'] = cname
data['企业类型'] = record['companyOrgType']
data['经营状态'] = record['regStatus']
data['法定代表人'] = record['legalPersonName']
data['注册资本'] = record['regCapital']
data['实缴资本'] = record['actualCapital']
data['经营范围'] = record['businessScope']
data['社会统一信用代码'] = record['creditCode']
data['成立日期'] = CommonUtils.time_stamp(record['estiblishTime'])
data['注册地址'] = record['regLocation']
data['登记机关'] = record['regInstitute']
data['核准日期'] = CommonUtils.time_stamp(record['approvedTime'])
data['更新时间'] = CommonUtils.get_current_time()
return data
else:
return False
record = record['基本信息']
data = dict()
data['企业ID'] = cid
data['企业名称'] = name
data['企业类型'] = record['companyOrgType']
data['经营状态'] = record['regStatus']
data['法定代表人'] = record['legalPersonName']
data['注册资本'] = record['regCapital']
data['实缴资本'] = record['actualCapital']
data['经营范围'] = record['businessScope']
data['社会统一信用代码'] = record['creditCode']
data['成立日期'] = CommonUtils.time_stamp(record['estiblishTime'])
data['注册地址'] = record['regLocation']
data['登记机关'] = record['regInstitute']
data['核准日期'] = CommonUtils.time_stamp(record['approvedTime'])
data['更新时间'] = CommonUtils.get_current_time()
if len(condition['cid']):
cid = condition['cid']
cname = DB_TEST.find_single_column(
'WR_DataBase_v2_test',
'C3.1_工商信息',
{'企业ID': cid},
'企业名称'
)
update = save_data(cname=cname, cid=cid)
if update:
DB_TEST.update_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业ID": cid},
update
)
return 'success'
else:
return 'update failed'
elif len(condition['cname']):
cname = condition['cname']
cid = CompanyUtils.make_new_cid()
insert = save_data(cname=cname, cid=cid)
if insert:
DB_TEST.upsert_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业名称": cname},
insert
)
return 'success'
else:
return '"new failed"'
else:
return 'parameter exception'
DB_TEST.upsert_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业名称": name},
data
)
@staticmethod
def edit_basic_information(**kwargs):
"""编辑工商信息"""
param = kwargs['param']
cid = param['cid']
body = param['body']
return '导入成功'
try:
business = BusinessInfo()
business.cname = body['企业名称']
business.ctype = body['企业类型']
business.status = body['经营状态']
business.representative = body['法定代表人']
business.registered_capital = body['注册资本']
business.paid_capital = body['实缴资本']
business.code = body['社会统一信用代码']
business.establishment_date = body['成立日期']
business.address = body['注册地址']
business.agencies = body['登记机关']
business.approved_date = body['核准日期']
business.business_scope = body['经营范围']
DB_TEST.update_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业ID": cid},
business.fields_toggle()
)
return 'success'
except Exception:
return 'edit failed'
@staticmethod
def view_basic_information(**kwargs):
@ -57,7 +123,16 @@ class BasicImpl(object):
[]
)
return record
if record:
result = {
"info": "success",
"result": record
}
else:
result = {
"info": "no result"
}
return result
def get_cid(self, **kwargs):
"""查询工商信息中的企业ID"""

48
Company/Basic/BasicRoutes.py
View File

@ -7,54 +7,6 @@ from Utils.RouteUtil import RouteParamsCheck
basic_route = Blueprint('basic', __name__)
@basic_route.route('/business_info/auto_update', methods=['GET'])
def basic_information_route():
"""工商信息-新增/更新（自动）"""
try:
RouteParamsCheck(req=request.args, params=["cname"]).required()
impl = BasicImpl()
result = impl.basic_information(name=request.args['cname'])
return {"info": "工商信息", "result": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@basic_route.route('/business_info/view', methods=['GET'])
def view_basic_information_route():
"""工商信息-查看"""
try:
RouteParamsCheck(req=request.args, params=["cid"]).required()
impl = BasicImpl()
result = impl.view_basic_information(cid=request.args['cid'])
return {"info": "工商信息", "result": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@basic_route.route('/shareholder/auto_update', methods=['GET'])
def shareholders_route():
"""股东信息-新增/更新（自动）"""
try:
RouteParamsCheck(req=request.args, params=["cname"]).required()
impl = BasicImpl()
result = impl.shareholders_information(name=request.args['cname'])
return {"info": "股东信息", "result": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@basic_route.route('/shareholder/view', methods=['GET'])
def view_shareholders_route():
"""股东信息-查看"""
try:
RouteParamsCheck(req=request.args, params=["cid"]).required()
impl = BasicImpl()
result = impl.view_shareholders(cid=request.args['cid'])
return {"info": "股东信息", "result": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@basic_route.route('/executive/auto_update', methods=['GET'])
def executive_information_route():
"""高管信息-新增/更新（自动）"""

135
Company/Basic/Business/BusinessImpl.py Normal file
View File

@ -0,0 +1,135 @@
from Company.Basic.Business.BusinessObj import BusinessInfo
from Company.CompanyUtils import CompanyUtils
from DBHelper.MongoHelperInstance import DB_TYC, DB_TEST
from Utils.CommonUtil import CommonUtils
class BusinessImpl(object):
"""工商信息"""
@staticmethod
def basic_information(**kwargs):
"""工商信息"""
condition = kwargs['condition']
def save_data(**args):
"""查询保存数据"""
company_name = args['cname']
company_id = args['cid']
record = DB_TYC.find_single_data(
'公司背景',
'基本信息',
{'企业名称': company_name},
['基本信息']
)
if record:
record = record['基本信息']
data = dict()
data['企业ID'] = company_id
data['企业名称'] = cname
data['企业类型'] = record['companyOrgType']
data['经营状态'] = record['regStatus']
data['法定代表人'] = record['legalPersonName']
data['注册资本'] = record['regCapital']
data['实缴资本'] = record['actualCapital']
data['经营范围'] = record['businessScope']
data['社会统一信用代码'] = record['creditCode']
data['成立日期'] = CommonUtils.time_stamp(record['estiblishTime'])
data['注册地址'] = record['regLocation']
data['登记机关'] = record['regInstitute']
data['核准日期'] = CommonUtils.time_stamp(record['approvedTime'])
data['更新时间'] = CommonUtils.get_current_time()
return data
else:
return False
if len(condition['cid']):
cid = condition['cid']
cname = DB_TEST.find_single_column(
'WR_DataBase_v2_test',
'C3.1_工商信息',
{'企业ID': cid},
'企业名称'
)
update = save_data(cname=cname, cid=cid)
if update:
DB_TEST.update_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业ID": cid},
update
)
return 'success'
else:
return 'update failed'
elif len(condition['cname']):
cname = condition['cname']
cid = CompanyUtils.make_new_cid()
insert = save_data(cname=cname, cid=cid)
if insert:
DB_TEST.upsert_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业名称": cname},
insert
)
return 'success'
else:
return '"new failed"'
else:
return 'parameter exception'
@staticmethod
def edit_basic_information(**kwargs):
"""编辑工商信息"""
param = kwargs['param']
cid = param['cid']
body = param['body']
try:
business = BusinessInfo()
business.cname = body['企业名称']
business.ctype = body['企业类型']
business.status = body['经营状态']
business.representative = body['法定代表人']
business.registered_capital = body['注册资本']
business.paid_capital = body['实缴资本']
business.code = body['社会统一信用代码']
business.establishment_date = body['成立日期']
business.address = body['注册地址']
business.agencies = body['登记机关']
business.approved_date = body['核准日期']
business.business_scope = body['经营范围']
DB_TEST.update_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业ID": cid},
business.fields_toggle()
)
return 'success'
except Exception:
return 'edit failed'
@staticmethod
def view_basic_information(**kwargs):
"""查看工商信息"""
cid = kwargs['cid']
record = DB_TEST.find_single_data(
"WR_DataBase_v2_test",
"C3.1_工商信息",
{"企业ID": cid},
[]
)
if record:
result = {
"info": "success",
"result": record
}
else:
result = {
"info": "no result"
}
return result

33
Company/Basic/Business/BusinessObj.py Normal file
View File

@ -0,0 +1,33 @@
from Utils.ObjUtil import SpecObject
from Utils.ValidateUtil import ValidateAttr, Validate
class BusinessInfo(SpecObject):
"""工商信息"""
cname = ValidateAttr(field='cname', type=str)
ctype = ValidateAttr(field='ctype', type=str)
status = ValidateAttr(field='status', type=str)
representative = ValidateAttr(field='representative', type=str)
registered_capital = ValidateAttr(field='registered_capital', type=str)
paid_capital = ValidateAttr(field='paid_capital', type=str)
code = ValidateAttr(field='code', type=str)
establishment_date = ValidateAttr(field='establishment_date', func=Validate.date_format)
address = ValidateAttr(field='address', type=str)
agencies = ValidateAttr(field='agencies', type=str)
approved_date = ValidateAttr(field='approved_date', func=Validate.date_format)
business_scope = ValidateAttr(field='business_scope', type=str)
fields_map = {
"cname": "企业名称",
"ctype": "企业类型",
"status": "经营状态",
"representative": "法定代表人",
"registered_capital": "注册资本",
"paid_capital": "实缴资本",
"code": "社会统一信用代码",
"establishment_date": "成立日期",
"address": "注册地址",
"agencies": "登记机关",
"approved_date": "核准日期",
"business_scope": "经营范围"
}

43
Company/Basic/Business/BusinessRoute.py Normal file
View File

@ -0,0 +1,43 @@
from flask import Blueprint, request
from Company.Basic.BasicImpl import BasicImpl
from Utils.ErrorUtil import APIReturnError
from Utils.RouteUtil import RouteParamsCheck
business_route = Blueprint('business', __name__)
@business_route.route('/auto_update', methods=['GET'])
def basic_information_route():
"""工商信息-新增/更新（自动）"""
try:
RouteParamsCheck(req=request.args, params=["cname", "cid"]).required()
impl = BasicImpl()
result = impl.basic_information(condition=request.args)
return {"info": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@business_route.route('/edit', methods=['POST'])
def edit_basic_information_route():
"""工商信息-编辑"""
try:
RouteParamsCheck(req=request.json, params=["cid", "body"]).required()
impl = BasicImpl()
result = impl.edit_basic_information(param=request.json)
return {"info": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@business_route.route('/view', methods=['GET'])
def view_basic_information_route():
"""工商信息-查看"""
try:
RouteParamsCheck(req=request.args, params=["cid"]).required()
impl = BasicImpl()
result = impl.view_basic_information(cid=request.args['cid'])
return result
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code

0
Company/Basic/Business/__init__.py Normal file
View File

166
Company/Basic/Shareholder/ShareholderImpl.py Normal file
View File

@ -0,0 +1,166 @@
from Company.Basic.Shareholder.ShareholderObj import ShareholderList
from Company.CompanyUtils import CompanyUtils
from DBHelper.MongoHelperInstance import DB_TYC, DB_TEST
from Utils.CommonUtil import CommonUtils
class ShareholderImpl(object):
"""股东信息"""
@staticmethod
def auto_update(**kwargs):
"""新增/更新（自动）"""
condition = kwargs['condition']
def save_data(**args):
"""查询保存数据"""
def search_cid(name):
"""根据股东名称查询cid"""
cid_record = DB_TEST.find_single_column(
'WR_DataBase_v2_test',
'C3.1_工商信息',
{'企业名称': name},
'企业名称'
)
if cid_record:
return cid_record
else:
return None
company_name = args['cname']
company_id = args['cid']
records = DB_TYC.find_single_column(
'公司背景',
'企业股东',
{'企业名称': company_name},
'企业股东'
)
if records:
records = records['result']
data = dict()
list_data = list()
for record in records:
dict_data = dict()
dict_data['股东ID'] = search_cid(record['name'])
dict_data['股东名称'] = record['name']
dict_data['股东类型'] = '公司' if record['type'] == 1 else '人' if record['type'] == 2 else '其他'
dict_data['股东性质'] = None
dict_data['持股比例'] = record['capital'][0]['percent']
dict_data['认缴出资额'] = record['capital'][0]['amomon']
dict_data['认缴出资日期'] = record['capital'][0]['time']
list_data.append(dict_data)
data['企业ID'] = company_id
data['股东列表'] = list_data
data['更新时间'] = CommonUtils.get_current_time()
return data
else:
return False
if len(condition['cid']):
cid = condition['cid']
cname = DB_TEST.find_single_column(
'WR_DataBase_v2_test',
'C3.2_股东信息',
{'企业ID': cid},
'企业名称'
)
update = save_data(cname=cname, cid=cid)
if update:
DB_TEST.update_single_data(
"WR_DataBase_v2_test",
"C3.2_股东信息",
{"企业ID": cid},
update
)
return 'success'
else:
return 'update failed'
elif len(condition['cname']):
cname = condition['cname']
cid = CompanyUtils.make_new_cid()
insert = save_data(cname=cname, cid=cid)
if insert:
DB_TEST.upsert_single_data(
"WR_DataBase_v2_test",
"C3.2_股东信息",
{"企业名称": cname},
insert
)
return 'success'
else:
return '"new failed"'
else:
return 'parameter exception'
@staticmethod
def delete(**kwargs):
"""删除"""
cid = kwargs['cid']
res = DB_TEST.delete_single_data(
"WR_DataBase_v2_test",
"C3.2_股东信息",
{"企业ID": cid}
)
if res:
result = {
"info": "success",
}
else:
result = {
"info": "no result"
}
return result
@staticmethod
def edit(**kwargs):
"""编辑"""
param = kwargs['param']
cid = param['cid']
body = param['body']
name = param['shareholder']
try:
shareholder = ShareholderList()
shareholder.sid = body['股东ID']
shareholder.name = body['股东名称']
shareholder.stype = body['股东类型']
shareholder.quality = body['股东性质']
shareholder.proportion = body['持股比例']
shareholder.amount = body['认缴出资额']
shareholder.date = body['认缴出资日期']
update_data = shareholder.fields_toggle()
DB_TEST.update_single_data_in_array(
"WR_DataBase_v2_test",
"C3.2_股东信息",
{"企业ID": cid},
{"股东名称": name},
update_data
)
return 'success'
except Exception:
return 'edit failed'
@staticmethod
def view(**kwargs):
"""查看"""
cid = kwargs['cid']
record = DB_TEST.find_single_data(
"WR_DataBase_v2_test",
"C3.2_股东信息",
{"企业ID": cid},
[]
)
if record:
result = {
"info": "success",
"result": record
}
else:
result = {
"info": "no result"
}
return result

36
Company/Basic/Shareholder/ShareholderObj.py Normal file
View File

@ -0,0 +1,36 @@
from Utils.ObjUtil import SpecObject
from Utils.ValidateUtil import ValidateAttr, Validate
class ShareholderList(SpecObject):
"""股东列表"""
sid = ValidateAttr(field='sid', type=str, default=None)
name = ValidateAttr(field='name', type=str)
stype = ValidateAttr(field='stype', type=str)
quality = ValidateAttr(field='quality', type=str, default=None)
proportion = ValidateAttr(field='proportion', type=str)
amount = ValidateAttr(field='amount', type=[str, float, int])
date = ValidateAttr(field='date', fun=Validate.date_format)
fields_map = {
"sid": "股东ID",
"name": "股东名称",
"stype": "股东类型",
"quality": "股东性质",
"proportion": "持股比例",
"amount": "认缴出资额",
"date": "认缴出资日期"
}
class ShareholderObj(SpecObject):
"""股东信息"""
cid = ValidateAttr(field='cid', type=str, length=8)
shareholder_list = ValidateAttr(field='shareholder_list', instance_list=ShareholderList)
update_time = ValidateAttr(field='update_time', func=Validate.time_format)
fields_map = {
"cid": "企业ID",
"shareholder_list": "股东列表",
"update_time": "更新时间"
}

55
Company/Basic/Shareholder/ShareholderRoute.py Normal file
View File

@ -0,0 +1,55 @@
from flask import Blueprint, request
from Company.Basic.Shareholder.ShareholderImpl import ShareholderImpl
from Utils.ErrorUtil import APIReturnError
from Utils.RouteUtil import RouteParamsCheck
shareholder_route = Blueprint('shareholder', __name__)
@shareholder_route.route('/auto_update', methods=['GET'])
def auto_update_route():
"""新增/更新（自动）"""
try:
RouteParamsCheck(req=request.args, params=["cname", "cid"]).required()
impl = ShareholderImpl()
result = impl.auto_update(condition=request.args)
return {"info": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@shareholder_route.route('/delete', methods=['GET'])
def delete_route():
"""删除"""
try:
RouteParamsCheck(req=request.args, params=["cid"]).required()
impl = ShareholderImpl()
result = impl.delete(cid=request.args['cid'])
return result
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@shareholder_route.route('/edit', methods=['POST'])
def edit_route():
"""编辑"""
try:
RouteParamsCheck(req=request.json, params=["cid", "shareholder", "body"]).required()
impl = ShareholderImpl()
result = impl.edit(param=request.json)
return {"info": result}
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code
@shareholder_route.route('/view', methods=['GET'])
def view_route():
"""查看"""
try:
RouteParamsCheck(req=request.args, params=["cid"]).required()
impl = ShareholderImpl()
result = impl.view(cid=request.args['cid'])
return result
except APIReturnError as e:
return {"info": e.__str__()}, e.status_code

0
Company/Basic/Shareholder/__init__.py Normal file
View File

15
DBHelper/MongoHelper.py
View File

@ -221,7 +221,7 @@ class MongoHelper:
except Exception:
pass
def update_single_data_by_pull(self, param1, param2, param3, param4):
def delete_single_data_in_array(self, param1, param2, param3, param4):
"""
更新单条数据
param1: str 数据库
@ -259,19 +259,6 @@ class MongoHelper:
collection.delete_one(param3)
return True
def delete_single_data_in_array(self, param1, param2, param3, param4):
"""
删除数组中的单条数据
param1: str 数据库
param2: str 数据表
param3: dict 一级字段查询条件
param4: dict 二级字段查询条件
return: Bool
"""
collection = self.client[param1][param2]
collection.update_one(param3, {"$pull": param4})
return True
def find_file(self, param1, param2, param3):
"""
读取一个文件

2
Modules/Functions/FinIndex/FinIndexImpl.py
View File

@ -262,7 +262,7 @@ class FinanceIndexImpl(object):
)
# 删除标签关联的函数
for tag in tags:
DB_TEST.update_single_data_by_pull(
DB_TEST.delete_single_data_in_array(
'模型数据',
'标签数据',
{'标签ID': tag['标签ID']},

2
Modules/Indicators/Tags/TagsImpl.py
View File

@ -244,7 +244,7 @@ class TagsImpl(object):
else:
_id = None
DB_TEST.update_single_data_by_pull(
DB_TEST.delete_single_data_in_array(
'模型数据',
'标签数据',
{'标签ID': tag_id},

2
Modules/Interface/InterfaceImpl.py
View File

@ -255,7 +255,7 @@ class InterfaceImpl(object):
)
# 删除标签关联的接口
for tag in tags:
DB_TEST.update_single_data_by_pull(
DB_TEST.delete_single_data_in_array(
'模型数据',
'标签数据',
{'标签ID': tag['标签ID']},

2
Modules/Models/Score/ScoreImpl.py
View File

@ -627,7 +627,7 @@ class ScoreModelImpl(object):
)
# 删除标签关联的函数
for tag in tags:
DB_TEST.update_single_data_by_pull(
DB_TEST.delete_single_data_in_array(
'模型数据',
'标签数据',
{'标签ID': tag['标签ID']},

2
Modules/Questionnaire/Questionnairelmpl.py
View File

@ -270,7 +270,7 @@ class QuestionnaireImpl(object):
)
# 删除标签关联的问题
for tag in tags:
DB_TEST.update_single_data_by_pull(
DB_TEST.delete_single_data_in_array(
'模型数据',
'标签数据',
{'标签ID': tag['标签ID']},

2
Modules/Questions/QuestionsImpl.py
View File

@ -515,7 +515,7 @@ class QuestionImpl(object):
)
# 删除标签关联的问题
for tag in tags:
DB_TEST.update_single_data_by_pull(
DB_TEST.delete_single_data_in_array(
'模型数据',
'标签数据',
{'标签ID': tag['标签ID']},

4
app.py
View File

@ -2,6 +2,8 @@ from flask import Flask
from flask_cors import *
from Company.Basic.BasicRoutes import basic_route
from Company.Basic.Business.BusinessRoute import business_route
from Company.Basic.Shareholder.ShareholderRoute import shareholder_route
from Company.Common.CommonRoutes import common_route
from Company.Ip.IpRoutes import ip_route
from Company.Judicial.JudicialRiskRoutes import judicial_risk_route
@ -34,6 +36,8 @@ app.register_blueprint(filters_route, url_prefix='/rating/data_filters')
app.register_blueprint(file_route, url_prefix='/rating/obj_file')
app.register_blueprint(tyc_route, url_prefix='/rating/tyc_api')
app.register_blueprint(basic_route, url_prefix='/data/manage/company/basic')
app.register_blueprint(business_route, url_prefix='/data/manage/company/basic/business_info')
app.register_blueprint(shareholder_route, url_prefix='/data/manage/company/basic/shareholder')
app.register_blueprint(ip_route, url_prefix='/data/manage/company/ip')
app.register_blueprint(judicial_risk_route, url_prefix='/data/manage/company/judicial_risk')
app.register_blueprint(operational_route, url_prefix='/data/manage/company/operational_risk')