파이썬 소켓통신 예제 + 문자열 처리 방법
- Dominant language
- Jupyter Notebook
- Stars
- 0
- Forks
- 0
- PR merge metrics
- No merged PRs in 30d
Description
````
# 소켓을 사용하기 위해서는 socket을 import해야 한다.
import socket, threading;
import requests
from bs4 import BeautifulSoup
import time
import urllib.request
from selenium.webdriver import Chrome
from selenium import webdriver
import re
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
import datetime as dt
import pandas as pd
# binder함수는 서버에서 accept가 되면 생성되는 socket 인스턴스를 통해 client로 부터 데이터를 받으면 echo형태로 재송신하는 메소드이다.
def quick_search_channel(key_word):
chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument('--headless')
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument("--single-process")
chrome_options.add_argument("--disable-dev-shm-usage")
path='./chromedriver'
driver = webdriver.Chrome(path, chrome_options=chrome_options)
search = "https://www.youtube.com/results?search_query="+key_word
driver.get(search)
time.sleep(0.2)
driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-search/div[1]/ytd-two-column-search-results-renderer/div/ytd-section-list-renderer/div[1]/div[2]/ytd-search-sub-menu-renderer/div[1]/div/ytd-toggle-button-renderer/a/tp-yt-paper-button/yt-formatted-string').click()
time.sleep(0.2)
print("체크 포인트2")
driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-search/div[1]/ytd-two-column-search-results-renderer/div/ytd-section-list-renderer/div[1]/div[2]/ytd-search-sub-menu-renderer/div[1]/iron-collapse/div/ytd-search-filter-group-renderer[2]/ytd-search-filter-renderer[2]/a/div/yt-formatted-string').click()
time.sleep(0.2)
if 1==1 :
ch= driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-search/div[1]/ytd-two-column-search-results-renderer/div/ytd-section-list-renderer/div[2]/ytd-item-section-renderer/div[3]/ytd-channel-renderer[1]')
print("체크 포인트")
# 채널 이름
ch_title = ch.find_element_by_xpath('div/div[2]/a/div[1]/ytd-channel-name/div/div/yt-formatted-string').text
# 채널 동영상 개수
try:
ch_movie = ch.find_element_by_xpath('div/div[2]/a/div[1]/div/span[3]').text
ch_movie = ch_movie.replace("동영상 ","").replace("개","").replace(",","").replace(" ","")
ch_movie = int(ch_movie)
except:
ch_movie = 0
# 채널 구독자 수
try:
ch_subs = ch.find_element_by_xpath('div/div[2]/a/div[1]/div/span[1]').text
print(ch_subs)
ch_sub = ch_subs.replace("구독자 ","").replace(" subscribers","")
if ch_sub.endswith("M"):
ch_sub = ch_sub.replace("M","")
ch_sub = float(ch_sub)*1000000
ch_sub = int(ch_sub)
elif ch_sub.endswith("K"):
ch_sub = ch_sub.replace("K","")
ch_sub = float(ch_sub)*1000
ch_sub = int(ch_sub)
elif ch_sub.endswith("B"):
ch_sub = ch_sub.replace("B","")
ch_sub = float(ch_sub)*1000000000
ch_sub = int(ch_sub)
else:
ch_sub = int(ch_sub)
except:
ch_sub = 0
print(ch_sub)
# 채널 링크
ch_link = ch.find_element_by_xpath('div/div[2]/a').get_attribute('href')
driver.get(ch_link)
# 배너 이미지 링크
ch_banner = driver.find_element_by_xpath('//*[@id="header"]/ytd-c4-tabbed-header-renderer').get_attribute('style')
ch_banner = ch_banner.replace("--yt-channel-banner:url(","").replace(")","").replace("\\","")
# 채널 sns 리스트 ex) [인스타 링크, 인스타 이미지, 트위치 링크. 트위치 이미지 ...]
ch_sns_lists = driver.find_elements_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-browse[1]/div[3]/ytd-c4-tabbed-header-renderer/tp-yt-app-header-layout/div/tp-yt-app-header/div[2]/div[2]/div/div[2]/div[2]/a')
ch_sns_list = []
for ch_sns in ch_sns_lists:
href = ch_sns.get_attribute('href')
try :
href = href.split("&q=")
href = href[1]
href = href.replace("%2F%3F","/?").replace("%3A",":").replace("%2F","/").replace("%3F",".").replace("%3D","=")
icon = ch_sns.find_element_by_xpath('yt-img-shadow/img').get_attribute('src')
ch_sns_list.append([href,icon])
except:
href = href[0]
href = href.replace("%2F%3F","/?").replace("%3A",":").replace("%2F","/").replace("%3F",".").replace("%3D","=")
icon = ch_sns.find_element_by_xpath('yt-img-shadow/img').get_attribute('src')
ch_sns_list.append([href,icon])
ch_profile_img = driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-browse[1]/div[3]/ytd-c4-tabbed-header-renderer/tp-yt-app-header-layout/div/tp-yt-app-header/div[2]/div[2]/div/div[1]/yt-img-shadow/img').get_attribute('src')
driver.get(ch_link+"/about")
ch_explanation = driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-browse[1]/ytd-two-column-browse-results-renderer/div[1]/ytd-section-list-renderer/div[2]/ytd-item-section-renderer/div[3]/ytd-channel-about-metadata-renderer/div[1]/div[1]/yt-formatted-string[2]').text
ch_birth = driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-browse[1]/ytd-two-column-browse-results-renderer/div[1]/ytd-section-list-renderer/div[2]/ytd-item-section-renderer/div[3]/ytd-channel-about-metadata-renderer/div[2]/yt-formatted-string[2]/span[2]').text
ch_birth = ch_birth.replace(". ","-").replace(".","")
ch_total_view = driver.find_element_by_xpath('/html/body/ytd-app/div/ytd-page-manager/ytd-browse[1]/ytd-two-column-browse-results-renderer/div[1]/ytd-section-list-renderer/div[2]/ytd-item-section-renderer/div[3]/ytd-channel-about-metadata-renderer/div[2]/yt-formatted-string[3]').text
ch_total_view = ch_total_view.replace("조회수 ","").replace(",","").replace("회","").replace('views',"")
ch_total_view = int(ch_total_view)
result = ch_title+"!@#"+str(ch_movie)+"!@#"+ch_link+"!@#"+ch_banner+"!@#"+ch_profile_img+"!@#"+ch_explanation+"!@#"+ch_birth+"!@#"+ch_subs
print(result)
return result
def binder(client_socket, addr):
# 커넥션이 되면 접속 주소가 나온다.
print('Connected by', addr);
try:
# 접속 상태에서는 클라이언트로 부터 받을 데이터를 무한 대기한다.
# 만약 접속이 끊기게 된다면 except가 발생해서 접속이 끊기게 된다.
while True:
# socket의 recv함수는 연결된 소켓으로부터 데이터를 받을 대기하는 함수입니다. 최초 4바이트를 대기합니다.
data = client_socket.recv(4);
# 최초 4바이트는 전송할 데이터의 크기이다. 그 크기는 little big 엔디언으로 byte에서 int형식으로 변환한다.
# C#의 BitConverter는 big엔디언으로 처리된다.
length = int.from_bytes(data, "little");
# 다시 데이터를 수신한다.
data = client_socket.recv(length);
# 수신된 데이터를 str형식으로 decode한다.
msg = data.decode();
# 수신된 메시지를 콘솔에 출력한다.
print('Received from', addr, msg)
ans="";
if msg != "":
print("크롤링 시작")
ans = quick_search_channel(msg)
print("크롤링 완료")
# 수신된 메시지 앞에 「echo:」 라는 메시지를 붙힌다.
print("답신 :",ans)
msg = ans
# 바이너리(byte)형식으로 변환한다.
data = msg.encode();
# 바이너리의 데이터 사이즈를 구한다.
length = len(data);
# 데이터 사이즈를 little 엔디언 형식으로 byte로 변환한 다음 전송한다.
client_socket.sendall(length.to_bytes(4, byteorder='little'));
# 데이터를 클라이언트로 전송한다.
client_socket.sendall(data);
except:
# 접속이 끊기면 except가 발생한다.
print("except : " , addr);
finally:
# 접속이 끊기면 socket 리소스를 닫는다.
client_socket.close();
# 소켓을 만든다.
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM);
# 소켓 레벨과 데이터 형태를 설정한다.
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1);
# 서버는 복수 ip를 사용하는 pc의 경우는 ip를 지정하고 그렇지 않으면 None이 아닌 ''로 설정한다.
# 포트는 pc내에서 비어있는 포트를 사용한다. cmd에서 netstat -an | find "LISTEN"으로 확인할 수 있다.
server_socket.bind(('', 9999));
# server 설정이 완료되면 listen를 시작한다.
server_socket.listen();
try:
# 서버는 여러 클라이언트를 상대하기 때문에 무한 루프를 사용한다.
while True:
# client로 접속이 발생하면 accept가 발생한다.
# 그럼 client 소켓과 addr(주소)를 튜플로 받는다.
client_socket, addr = server_socket.accept();
th = threading.Thread(target=binder, args = (client_socket,addr));
# 쓰레드를 이용해서 client 접속 대기를 만들고 다시 accept로 넘어가서 다른 client를 대기한다.
th.start();
except:
print("server");
finally:
# 에러가 발생하면 서버 소켓을 닫는다.
server_socket.close();
````
Contributor guide
No contributing guide indexed for this repository
Research direction
The issue body contains the Python entry points quick_search_channel, binder, and the server socket setup, but names no repository file or test. Start by locating where this example belongs and clarify the intended string-processing or socket behavior; done requires an agreed example scope and verification steps.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- pandas, python
- Domain
- documentation
- Issue type
- Documentation
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 20/100