spider/get_company_message.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "08f2afc3",
   "metadata": {},
   "outputs": [],
   "source": [
    "from bs4 import BeautifulSoup\n",
    "import requests\n",
    "import time\n",
    "import pandas as pd\n",
    "from requests.auth import HTTPBasicAuth\n",
    "import os\n",
    "import xlwings as xw\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9b2a027c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 获取特定公司的指定信息\n",
    "def get_company_message(company, afterLogin_headers):\n",
    "    # 获取查询到的网页内容（全部）\n",
    "    search = sess.get('https://www.qcc.com/web/search?key={}'.format(company),headers=afterLogin_headers,timeout=10)\n",
    "    search.raise_for_status()\n",
    "    search.encoding = 'utf-8' #linux utf-8\n",
    "    soup = BeautifulSoup(search.text,features=\"html.parser\")\n",
    "    href = soup.find_all('a',{'class': 'title'})[0].get('href')\n",
    "    target_company = soup.find_all('a',{'class': 'title'})[0].text\n",
    "    time.sleep(4)\n",
    "    details = sess.get(href,headers=afterLogin_headers,timeout=10)\n",
    "    details.raise_for_status()\n",
    "    details.encoding = 'utf-8' #linux utf-8\n",
    "    details_soup = BeautifulSoup(details.text,features=\"html.parser\")\n",
    "    # 获取企业主页\n",
    "    conpany_homePage = details_soup.find_all('div',{'class': 'content'})[0]\n",
    "    # 获取工商信息\n",
    "    business_message = details_soup.find_all({'table': 'ntable'})[0].text\n",
    "    # 获取股东信息\n",
    "    try:\n",
    "        partner_message = details_soup.find_all(id=re.compile('partner'))[0].select('table')\n",
    "        # 如果列表为空，则代表改信息只有vip账号才能获取，设置为0跳过处理\n",
    "        if partner_message==[]:\n",
    "            partner_message = 0\n",
    "    except:\n",
    "        partner_message = 0 \n",
    "    # 获取主要人员\n",
    "    try:\n",
    "        main_people = details_soup.find_all(id='mainmember')[0].select('table')\n",
    "        # 如果列表为空，则代表改信息只有vip账号才能获取，设置为0跳过处理\n",
    "        if main_people==[]:\n",
    "            main_people = 0\n",
    "    except:\n",
    "        main_people = 0\n",
    "    # 获取对外投资\n",
    "    try:\n",
    "        external_touzi = details_soup.find_all(id=re.compile('touzilist'))[0].select('table')\n",
    "        # 如果列表为空，则代表改信息只有vip账号才能获取，设置为0跳过处理\n",
    "        if external_touzi==[]:\n",
    "            external_touzi = 0\n",
    "    except:\n",
    "        external_touzi = 0\n",
    "    # 获取变更记录\n",
    "    try:\n",
    "        change_record = details_soup.find_all(id=re.compile('changelist'))[0].select('table')\n",
    "        if change_record == []:\n",
    "            change_record = 0\n",
    "    except:\n",
    "        change_record = 0\n",
    "    \n",
    "    return [target_company,conpany_homePage, business_message, partner_message, main_people, external_touzi, change_record]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "899ef4c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 获取企业主页头部的信息\n",
    "def conpany_homePage_to_df(company_name,company_message):\n",
    "    if company_message==0:\n",
    "        homePage_df = pd.DataFrame()\n",
    "    else:\n",
    "        company_content = company_message.find_all('div', {'class': 'contact-info'})[0].text.replace(\"\\n\", \"\").replace(\" \", \"\")\n",
    "        # 法定代表人\n",
    "        try:\n",
    "            c1 = company_content.split(\"法定代表人：\")[1].split('关联')[0]\n",
    "        except:\n",
    "            c1 = '无'\n",
    "        # 统一社会信用代码\n",
    "        try:\n",
    "            c2 = company_content.split(\"统一社会信用代码：\")[1].split(\"复制\")[0]\n",
    "        except:\n",
    "            c2 = '无'\n",
    "        # 电话\n",
    "        c3 = company_content.split(\"电话：\")[1].split(\"同电话\")[0].split(\"更多\")[0]\n",
    "        # 官网\n",
    "        c4 = company_content.split(\"官网：\")[1].split(\"邮箱\")[0]\n",
    "        # 邮箱\n",
    "        c5 = company_content.split(\"邮箱：\")[1].split(\"复制\")[0]\n",
    "        # 地址\n",
    "        c6 = company_content.split(\"地址：\")[1].split(\"附近企业\")[0]\n",
    "        # 简介\n",
    "        c7 = company_content.split(\"简介：\")[1].split(\"复制\")[0]\n",
    "\n",
    "        homePage_df = pd.DataFrame({'公司名称': [company_name],'法定代表人': [c1],'统一社会信用代码': [c2],\\\n",
    "            '电话': [c3], '官网': [c4], '邮箱': [c5], '地址': [c6], '简介': [c7]})\n",
    "        \n",
    "    return homePage_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c8f31b65",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 获取工商信息的卡片\n",
    "def business_message_to_df(message):\n",
    "    # 统一社会信用代码\n",
    "    unified_social_credit_code = []\n",
    "    try:\n",
    "        unified_social_credit_code.append(message.split('统一社会信用代码')[1].split('复制')[0].replace(\" \",\"\").replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        unified_social_credit_code.append('无法收集')\n",
    "    # 企业名称\n",
    "    list_companys = []\n",
    "    try:\n",
    "        list_companys.append(message.split('企业名称')[1].split('复制')[0].replace(\" \",\"\").replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        list_companys.append('无法收集')\n",
    "    # 法定代表人\n",
    "    Legal_Person = []\n",
    "    try:\n",
    "        Legal_Person.append(message.split('法定代表人')[1].split('关联')[0].replace(\"\\n\",\"\").replace(\" \",\"\"))\n",
    "    except:\n",
    "        Legal_Person.append('无法收集')\n",
    "    # 登记状态\n",
    "    Registration_status = []\n",
    "    try:\n",
    "        Registration_status.append(message.split('登记状态')[1].split('成立日期')[0].replace(\" \",\"\").replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Registration_status.append('无法收集')\n",
    "    # 成立日期\n",
    "    Date_of_Establishment = []\n",
    "    try:\n",
    "        Date_of_Establishment.append(message.split('成立日期')[1].split('注册资本')[0].replace(\" \",\"\").replace(\"\\n\",\"\"))\n",
    "    except:  \n",
    "        Date_of_Establishment.append('无法收集')\n",
    "    # 注册资本\n",
    "    registered_capital = []\n",
    "    try:\n",
    "        registered_capital.append(message.split('注册资本')[1].split('实缴资本')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        registered_capital.append('无法收集')\n",
    "    # 实缴资本\n",
    "    contributed_capital = []\n",
    "    try:\n",
    "        contributed_capital.append(message.split('实缴资本')[1].split('核准日期')[0].replace(' ','').replace('\\n',''))\n",
    "    except:\n",
    "        contributed_capital.append('无法收集')\n",
    "    # 核准日期\n",
    "    Approved_date = []\n",
    "    try:\n",
    "        Approved_date.append(message.split('核准日期')[1].split('组织机构代码')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Approved_date.append('无法收集')\n",
    "    # 组织机构代码\n",
    "    Organization_Code = []\n",
    "    try:\n",
    "        Organization_Code.append(message.split('组织机构代码')[1].split('复制')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Organization_Code.append('无法收集')\n",
    "    # 工商注册号\n",
    "    companyNo = []\n",
    "    try:\n",
    "        companyNo.append(message.split('工商注册号')[1].split('复制')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        companyNo.append('无法收集')\n",
    "    # 纳税人识别号\n",
    "    Taxpayer_Identification_Number = []\n",
    "    try:\n",
    "        Taxpayer_Identification_Number.append(message.split('纳税人识别号')[1].split('复制')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Taxpayer_Identification_Number.append('无法收集')\n",
    "    # 企业类型\n",
    "    enterprise_type = []\n",
    "    try:\n",
    "        enterprise_type.append(message.split('企业类型')[1].split('营业期限')[0].replace('\\n','').replace(' ',''))\n",
    "    except:\n",
    "        enterprise_type.append('无法收集')\n",
    "    # 营业期限\n",
    "    Business_Term = []\n",
    "    try:\n",
    "        Business_Term.append(message.split('营业期限')[1].split('纳税人资质')[0].replace('\\n','').replace(' ',''))\n",
    "    except:\n",
    "        Business_Term.append('无法收集')\n",
    "    # 纳税人资质\n",
    "    Taxpayer_aptitude = []\n",
    "    try:\n",
    "        Taxpayer_aptitude.append(message.split('纳税人资质')[1].split('所属行业')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Taxpayer_aptitude.append('无法收集')\n",
    "    # 所属行业\n",
    "    sub_Industry = []\n",
    "    try:\n",
    "        sub_Industry.append(message.split('所属行业')[1].split('所属地区')[0].replace('\\n','').replace(' ',''))\n",
    "    except:\n",
    "        sub_Industry.append('无法收集')   \n",
    "    # 所属地区\n",
    "    sub_area = []\n",
    "    try:\n",
    "        sub_area.append(message.split('所属地区')[1].split('登记机关')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        sub_area.append('无法收集')\n",
    "    # 登记机关\n",
    "    Registration_Authority = []\n",
    "    try:\n",
    "        Registration_Authority.append(message.split('登记机关')[1].split('人员规模')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Registration_Authority.append('无法收集')\n",
    "    # 人员规模\n",
    "    staff_size = []\n",
    "    try:\n",
    "        staff_size.append(message.split('人员规模')[1].split('参保人数')[0].replace(' ','').replace('\\n',''))\n",
    "    except:\n",
    "        staff_size.append('无法收集')\n",
    "    # 参保人数\n",
    "    Number_of_participants = []\n",
    "    try:\n",
    "        Number_of_participants.append(message.split('参保人数')[1].split('趋势图')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Number_of_participants.append('无法收集')\n",
    "    # 曾用名\n",
    "    Used_Name = []\n",
    "    try:\n",
    "        Used_Name.append(message.split('曾用名')[1].split('英文名')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Used_Name.append('无法收集')\n",
    "    # 英文名\n",
    "    English_name = []\n",
    "    try:\n",
    "        English_name.append(message.split('英文名')[1].split('进出口企业代码')[0].replace('\\n','').replace(' ',''))\n",
    "    except:\n",
    "        English_name.append('无法收集')\n",
    "    # 进出口企业代码\n",
    "    import_and_export_code = []\n",
    "    try:\n",
    "        import_and_export_code.append(message.split('进出口企业代码')[1].split('复制')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        import_and_export_code.append('无法收集')\n",
    "    # 注册地址\n",
    "    register_adress = []\n",
    "    try:\n",
    "        register_adress.append(message.split('注册地址')[1].split('附近企业')[0].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        register_adress.append('无法收集')\n",
    "    # 经营范围\n",
    "    Business_Scope = []\n",
    "    try:\n",
    "        Business_Scope.append(message.split('经营范围')[1].replace(' ','').replace(\"\\n\",\"\"))\n",
    "    except:\n",
    "        Business_Scope.append('无法收集')\n",
    "    df = pd.DataFrame({'统一社会信用代码': unified_social_credit_code,\\\n",
    "                       '企业名称': list_companys,\\\n",
    "                       '法定代表人':Legal_Person,\\\n",
    "                      '登记状态':Registration_status,\\\n",
    "                      '成立日期':Date_of_Establishment,\\\n",
    "                      '注册资本':registered_capital,\\\n",
    "                      '实缴资本':contributed_capital,\\\n",
    "                      '核准日期':Approved_date,\\\n",
    "                      '组织机构代码':Organization_Code,\\\n",
    "                      '工商注册号':companyNo,\\\n",
    "                      '纳税人识别号':Taxpayer_Identification_Number,\\\n",
    "                      '企业类型':enterprise_type,\\\n",
    "                      '营业期限':Business_Term,\\\n",
    "                      '纳税人资质':Taxpayer_aptitude,\n",
    "                      '所属行业':sub_Industry,\\\n",
    "                      '所属地区':sub_area,\\\n",
    "                      '登记机关':Registration_Authority,\\\n",
    "                      '人员规模':staff_size,\\\n",
    "                      '参保人数':Number_of_participants,\\\n",
    "                      '曾用名': Used_Name, \\\n",
    "                      '英文名':English_name, \\\n",
    "                      '进出口企业代码': import_and_export_code, \\\n",
    "                      '注册地址':register_adress,\\\n",
    "                      '经营范围':Business_Scope})\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dfeec0c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 获取列名为横向的卡片信息\n",
    "def col_is_vertical_to_df(company_name,message):\n",
    "    if message==0:\n",
    "        col_df = pd.DataFrame()\n",
    "    else:\n",
    "        # 定位到你要的那个表格，靠id=特定id，然后再在这个基础上找到table标签\n",
    "        list_col = []\n",
    "        list_row_all = []\n",
    "        # 获取列名\n",
    "        col_name = message[0].select('tr')[0].select('th')\n",
    "        for i in range(len(col_name)):\n",
    "            list_col.append(col_name[i].text.replace(' ','').replace('\\n',''))\n",
    "        # 获取每一行的信息\n",
    "        row = len(message[0].select('tr'))-1\n",
    "        for i in range(row):\n",
    "            list_row = []\n",
    "            col_i = message[0].select('tr')[i+1].select('td')\n",
    "            for i in range(len(col_i)):\n",
    "                list_row.append(col_i[i].text.replace(' ','').replace('\\n','').split('关联')[0])\n",
    "            list_row_all.append(list_row)\n",
    "        # 汇总为dataframe\n",
    "        col_df = pd.DataFrame(list_row_all, columns=list_col)\n",
    "        col_df['公司名称'] = company_name\n",
    "    return col_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e60bea74",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 对于有规则的固定列，授予专门的处理函数\n",
    "def regular_line(app, sheet_name, df):\n",
    "    if len(df) == 0:\n",
    "        pass\n",
    "    else:\n",
    "        # 给对应工作表添加内容\n",
    "        sheet = app.sheets[sheet_name]\n",
    "\n",
    "        # 获取工作表目前的行\n",
    "        if sheet.used_range.rows.count == 1:\n",
    "            now_row = sheet.used_range.rows.count\n",
    "            # 将dataframe的数据写进xlsx\n",
    "            sheet.range('A' + str(now_row)).options(pd.DataFrame, expand='table', index=False).value = df\n",
    "            sheet.range('A1').expand('right').api.Font.Bold = True\n",
    "        else:\n",
    "            now_row = sheet.used_range.rows.count + 1\n",
    "            # 将dataframe的数据写进xlsx\n",
    "            sheet.range('A' + str(now_row)).options(pd.DataFrame, expand='table',header=False,index=False).value = df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e0bcc43",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 对于无规则的固定列，授予专门的处理函数\n",
    "def Irregular_line(app, sheet_name, df):\n",
    "    if len(df) == 0:\n",
    "        pass\n",
    "    else:\n",
    "        # 给对应工作表添加内容\n",
    "        sheet = app.sheets[sheet_name]\n",
    "\n",
    "        # 获取工作表目前的行\n",
    "        if sheet.used_range.rows.count == 1:\n",
    "            now_row = sheet.used_range.rows.count\n",
    "        else:\n",
    "            now_row = sheet.used_range.rows.count + 2\n",
    "        # 将dataframe的数据写进xlsx\n",
    "        sheet.range('A' + str(now_row)).options(pd.DataFrame, expand='table', index=False).value = df\n",
    "\n",
    "        # 将列名那一行上黄色，然后数据行上灰色\n",
    "        for i in range(len(df2)+1):\n",
    "            if i == 0:\n",
    "                sheet.range('A' + str(now_row)).expand('right').color = (255,255,0)\n",
    "            else:\n",
    "                sheet.range('A' + str(now_row+i)).expand('right').api.Font.Color = 0x00000 \n",
    "                sheet.range('A' + str(now_row+i)).expand('right').color = (220,220,220)\n",
    "                print('A' + str(now_row+i))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7d5db106",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 对于因为图片带有姓的原因造成姓重复的情况，进行去重，此函数只适合于大部分情况\n",
    "def dedup_name(origin_name):\n",
    "    if len(origin_name)>=2:\n",
    "        if(origin_name[0:1] == origin_name[1:2]):\n",
    "            target_name = origin_name[1:]\n",
    "        else:\n",
    "            target_name = origin_name\n",
    "    target_name = target_name.replace('最终受益人','').replace('实际控制人','').replace('大股东','').replace('有限制','').replace('高消费','')\n",
    "    \n",
    "    return target_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bda70dd6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 修改这些参数即可运用本案例\n",
    "user_agent = '此代码上方介绍了获取的方法'\n",
    "cookie = '此代码上方介绍了获取的方法'\n",
    "user_name = '企查查登录成功后的用户名'\n",
    "password = '登录成功后的企查查密码'\n",
    "save_path = '自己目录的绝对路径/某某.csv'\n",
    "\n",
    "# 测试所用\n",
    "companys = ['深圳市腾讯计算机系统有限公司', '阿里巴巴（中国）有限公司']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0b5f681c",
   "metadata": {},
   "outputs": [],
   "source": [
    "if __name__ == '__main__':\n",
    "    # 保持会话\n",
    "    # 新建一个session对象\n",
    "    sess = requests.session()\n",
    "\n",
    "    # 添加headers（header为自己登录的企查查网址，输入账号密码登录之后所显示的header，此代码的上方介绍了获取方法）\n",
    "    afterLogin_headers = {'User-Agent': user_agent,'cookie':cookie}\n",
    "\n",
    "    # 以用户的身份认证,方便后面执行查询指令\n",
    "    res = requests.get('https://www.qcc.com/',headers=afterLogin_headers, auth=HTTPBasicAuth(username, password))\n",
    "    print(res.status_code)\n",
    "    \n",
    "    # 获取列表中的每一个公司的信息\n",
    "    app = xw.App(visible=False)\n",
    "    try:\n",
    "        for i in range(len(companys)):\n",
    "\n",
    "            # 获取该公司的部分网页信息\n",
    "            company_message = get_company_message(companys[i],afterLogin_headers)\n",
    "            # 企业主页 \n",
    "            df1 = conpany_homePage_to_df(company_message[0],company_message[1])\n",
    "            # 工商信息 \n",
    "            df2 = business_message_to_df(company_message[2])\n",
    "            df2['法定代表人'] = df2['法定代表人'].apply(dedup_name)\n",
    "            # 股东信息 \n",
    "            df3 = col_is_vertical_to_df(company_message[0],company_message[3])\n",
    "            # 主要人员\n",
    "            df4 = col_is_vertical_to_df(company_message[0],company_message[4])\n",
    "            df4['姓名'] = df4['姓名'].apply(dedup_name)\n",
    "            # 对外投资\n",
    "            df5 = col_is_vertical_to_df(company_message[0],company_message[5])\n",
    "            # 变更记录\n",
    "            df6 = col_is_vertical_to_df(company_message[0],company_message[6])\n",
    "\n",
    "            # 检查目标xlsx是否存在，不存在则提前创建好\n",
    "            if not os.path.exists(save_path):\n",
    "                workbook = app.books.add()\n",
    "                workbook.save(save_path)\n",
    "                workbook.close()\n",
    "                time.sleep(1)\n",
    "\n",
    "                # 刚创建的时候，是空文件，重命名或新增对应工作表，进行第一次写入\n",
    "                wb = app.books.open(save_path)\n",
    "                wb.sheets[0].name = wb.sheets[0].name.replace('Sheet1','企业主页')\n",
    "                wb.sheets.add('工商信息',after='企业主页')\n",
    "                wb.sheets.add('股东信息',after='工商信息')\n",
    "                wb.sheets.add('主要人员',after='股东信息')\n",
    "                wb.sheets.add('对外投资',after='主要人员')\n",
    "                wb.sheets.add('变更记录',after='对外投资')\n",
    "                wb.save()\n",
    "                wb.close()\n",
    "\n",
    "            # 写数据进xlxs\n",
    "            wb_write = app.books.open(save_path)\n",
    "            regular_line(wb_write, '企业主页', df1)\n",
    "            regular_line(wb_write, '工商信息', df2)\n",
    "            Irregular_line(wb_write, '股东信息', df3)\n",
    "            Irregular_line(wb_write, '主要人员', df4)\n",
    "            Irregular_line(wb_write, '对外投资', df5)\n",
    "            Irregular_line(wb_write, '变更记录', df6)\n",
    "            wb_write.save()\n",
    "            wb_write.close()\n",
    "            time.sleep(1)\n",
    "\n",
    "        app.quit()\n",
    "    except Exception as e:\n",
    "        app.quit()\n",
    "        print(e)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.10.1 64-bit",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.1"
  },
  "vscode": {
   "interpreter": {
    "hash": "a42ccb73e7d9bfdf27e036f1d2b8b681e55fc0743cc5586bc2474d4a60f4b886"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}