# -*- coding: UTF-8 -*- import os import json import requests from urllib.parse import urlparse, parse_qs host = 'https://backend.igengmei.com/' split_s = '###' feed_cards_fn = lambda j: j.get('data', {}).get('features', []) DYNAMIC_PARSE_APIS = { 'api/index/v8?tabtype=choice': feed_cards_fn, 'api/index/v8?tabtype=home_video': feed_cards_fn, } DYN_API_CONF_NAME = 'dynamic_apis.conf' EMPTY_RESP = """ { "error_extra": null, "extra": null, "user_type": {}, "error": 0, "message": "", "data": "" } """ STATIC_JSON_DATA = { 'api/live/enter': EMPTY_RESP, 'api/janus/live/polling': """ { "error": 0, "message": "", "extra": null, "error_code": null, "error_extra": null, "data": { "delay": 10, "ball_info": null, "banner_info": null, "banner_info_second": null, "stream_status": { "is_finish": false }, "pop_gift": null, "pop_service": null }, "user_type": {} } """, 'api/janus/live/m/enter': EMPTY_RESP, 'api/janus/live/m/polling': """ { "error": 0, "message": "", "extra": null, "error_code": null, "error_extra": null, "data": { "delay": 10, "ball_info": null, "banner_info": null, "banner_info_second": null, "stream_status": { "is_finish": false }, "pop_gift": null, "pop_service": null }, "user_type": {} } """, 'api/janus/live/m/leave': EMPTY_RESP, 'api/user/follow/add': """ { "error": 0, "message": "\u5173\u6ce8\u6210\u529f", "extra": { "general_additional_data": { "sub_title": "\u7b2c\u4e00\u65f6\u95f4\u83b7\u5f97TA\u7684\u6700\u65b0\u6d88\u606f\uff1f", "general_key": "notification_alert", "type": "attention_click_do", "title": "\u6253\u5f00\u63a8\u9001\u901a\u77e5" } } } """, 'api/user/follow/del': """ { "message": "\u53d6\u6d88\u5173\u6ce8\u6210\u529f", "extra": {}, "error": 0 } """, } def do_request(url): if '?' in url: join_symbol = '&' else: join_symbol = '?' url = '{}{}version=7.37.0'.format(url, join_symbol) resp = requests.get(url) return resp.json() def gen_uri_data_to_json(uri, jname='', only_get=False): print(uri, jname) if not uri and not jname: return query = '' json_path_and_name = uri if '?' in uri: json_path_and_name, query = uri.split('?') if jname and query: json_file_name = '{}.json'.format(jname) json_file_dir = json_path_and_name else: json_file_name = '{}.json'.format(os.path.basename(json_path_and_name)) json_file_dir = os.path.dirname(json_path_and_name) json_file_path = '{}/{}'.format(json_file_dir, json_file_name) if uri in STATIC_JSON_DATA: _data = STATIC_JSON_DATA[uri] json_data = json.loads(_data) else: url = '{}{}'.format(host, uri) json_data = do_request(url) if not only_get: os.makedirs(json_file_dir, exist_ok=True) with open(json_file_path, 'w') as pf: pf.write(json.dumps(json_data, indent=4)) return json_data def gen_nginx_location_rule(uri, jname=''): pass def get_val_from_query(qs, key, default=''): vals = qs.get(key) if not vals: return default return vals[0] def parse_api_from_card(card_data): DEFAULT_RET = None if 'card_name' not in card_data: return DEFAULT_RET gmurl = card_data.get(card_data['card_name'], {}).get('gm_url', '') if not gmurl: return DEFAULT_RET _url_s = urlparse(gmurl) if _url_s.scheme != 'gengmei' or not _url_s.netloc: return DEFAULT_RET _qs = parse_qs(_url_s.query) if _url_s.netloc == 'tractate': # 帖子 obj_id = get_val_from_query(_qs, 'tractate_id') if not obj_id: return DEFAULT_RET ret = [ 'api/tractate/new_detail?tractate_id={}###{}'.format(obj_id, obj_id), 'api/tractate/related_service?tractate_id={}###{}'.format(obj_id, obj_id), 'api/tractate/reply?id={}###{}'.format(obj_id, obj_id), 'api/personal/recommends/v1?id={}&from_type=tractate###{}_{}'.format(obj_id, 'tractate', obj_id), ] return ret elif _url_s.netloc == 'diary': # 日记本 obj_id = get_val_from_query(_qs, 'diary_id') if not obj_id: return DEFAULT_RET # import pdb; pdb.set_trace() diary_info = gen_uri_data_to_json('api/diary/get/v2?diary_id={}'.format(obj_id), str(obj_id), only_get=True) diary_service_id = diary_info.get('data', {}).get('diary_info', {}).get('service_id', '') ret = [ 'api/diary/get/v2?diary_id={}###{}'.format(obj_id, obj_id), 'api/diary/services?diary_id={}&service_id={}###{}'.format(obj_id, diary_service_id, obj_id), 'api/personal/recommends/v1?id={}&from_type=diary###{}_{}'.format(obj_id, 'diary', obj_id), 'api/diary/topic/{}/_data'.format(obj_id), ] return ret elif _url_s.netloc == 'question_detail': # 问答 obj_id = get_val_from_query(_qs, 'question_id') if not obj_id: return DEFAULT_RET ret = [ 'api/answer/detail?answer_id={}###{}'.format(obj_id, obj_id), 'hybrid/question_v1/{}###{}'.format(obj_id, obj_id), 'hybrid/question/new_answer_list/_data?question_id={}###{}'.format(obj_id, obj_id), 'api/personal/recommends/v1?id={}&from_type=question###{}_{}'.format(obj_id, 'question', obj_id), ] return ret elif _url_s.netloc == 'live_player': # 直播 obj_data = card_data.get(card_data['card_name'], {}) stream_id = obj_data.get('stream_id', '') channel_id = obj_data.get('channel_id', '') if not stream_id or not channel_id: return DEFAULT_RET ret = [ 'api/live/get_live_info/audience?channel_id={}###{}'.format(channel_id, channel_id), 'api/janus/live/live_gifts?gift_type=1&stream_id={}&channel_id={}###{}_{}'.format(stream_id, channel_id, stream_id, channel_id), 'api/janus/live/audience_services?stream_id={}&channel_id={}###{}_{}'.format(stream_id, channel_id, stream_id, channel_id), 'api/janus/live/m/get_live_info/audience?channel_id={}###{}'.format(channel_id, channel_id), 'api/janus/live/m/live_gifts?gift_type=1&stream_id={}&channel_id={}###{}_{}'.format(stream_id, channel_id, stream_id, channel_id), ] return ret elif _url_s.netloc == 'service': # 美购 obj_id = get_val_from_query(_qs, 'service_id') if not obj_id: return DEFAULT_RET ret = [] return ret else: return DEFAULT_RET def parse_cards(cards): print('parse_cards', len(cards)) with open(DYN_API_CONF_NAME, 'a') as pf: for card_data in cards: api_confs = parse_api_from_card(card_data) if not api_confs: continue for api_conf in api_confs: pf.write('{}\n'.format(api_conf)) def _conf_line_parse(line): line = line.strip() if line.startswith('#'): return '', '' if split_s in line: uri, jname = line.split('###') else: uri, jname = line, '' return uri, jname def base_api_json_generate(): for line in open('apis.conf'): uri, jname = _conf_line_parse(line) json_data = gen_uri_data_to_json(uri, jname) # 根据feed生成动态json数据 if uri in DYNAMIC_PARSE_APIS: parse_fn = DYNAMIC_PARSE_APIS[uri] cards = parse_fn(json_data) parse_cards(cards) def dyn_api_json_generate(): if not os.path.exists(DYN_API_CONF_NAME): print('{} not exists'.format(DYN_API_CONF_NAME)) return for line in open(DYN_API_CONF_NAME): uri, jname = _conf_line_parse(line) gen_uri_data_to_json(uri, jname) def clean_dyn_conf(): if os.path.exists(DYN_API_CONF_NAME): os.remove(DYN_API_CONF_NAME) def main(): clean_dyn_conf() base_api_json_generate() dyn_api_json_generate() if __name__ == '__main__': main()