[Python] snscrape 패키지를 이용하여 id / 기간별 특정 키워드가 포함된 트위터 자료 수집

 정보

• 업무명     : snscrape 패키지를 이용하여 id / 기간별 특정 키워드가 포함된 트위터 자료 수집
• 작성자     : 박진만
• 작성일     : 2021-02-03
• 설   명      :
• 수정이력 :

 

 내용

[특징]

• snscrape 패키지를 이용하여 트위터에 접근하고 원하는 자료를 수집한다.

 

[기능]

• 트위터 자료 수집
• 특정 기간 / 아이디에 대해 원하는 트위터 자료 선택 후 저장 및 하기의 작업을 수행
  • 1. 기간 / 아이디에 따른 트위터 수집
  • 2. 리트윗 / 리플라이 제거 (선택사항)
  • 3. 키워드가 일치하는 트윗만 선택 (선택사항)
  • 4. 출력및 로그 저장

 

[활용 자료]

• 없음

 

[자료 처리 방안 및 활용 분석 기법]

• 없음

 

[사용법]

• 소스 코드 예시 참조

 

[사용 OS]

• Windows 10

 

[사용 언어]

• Python 3.8
• snscrape

 

 소스 코드

[준비]

• 라이브러리 로드
  • snscrape 패키지를 포함한 데이터 처리 라이브러리 로드
    
  • 본문에는 경고를 무시하는 코드를 추가로 삽입함
  • 패키지 설치가 되지 않았다면 pip 명령어를 통해 간단하게 설치가 가능함

# 패키지 읽기 
import twitter
import numpy as np
import pandas as pd
import string
import re
import warnings
import os
import snscrape
import snscrape.modules.twitter as sntwitter
from time import sleep
import datetime
 
warnings.filterwarnings(action='ignore')

 

[함수 생성]

• 트위터 자료를 수집하는 함수 생성
  • 트위터 아이디 / 찾는 시작 날짜 / 종료 날짜를 각각 인자로 넣음
  • 트위터 수집 후 출력되는 데이터 프레임을 반환함
  • 특정 아이디 / 기간 에 대해서만 검색 하고자 하는 경우 해당 함수를 태우는 것으로 작업을 종료 해도 무방함

# 인자 : 트위터 계정명,시작날짜(yyyy-mm-dd), 종료날짜(yyyy-mm-dd)
# 리턴되는 결과 : 해당 계정의 트위터 결과 데이터 프레임 (url,time,id,content,text,username)
# 계정 이름이 누락되는 경우 데이터 프레임이 반환되지 않고 대신 String 이 반환됨
def read_tweet_list(twitterName,startDay,endDay):
    
    tweets_list1 = []
    tweets_df2 = pd.DataFrame(columns=['URL','Datetime', 'Tweet Id','Content', 'Username'])
    
    # 계정 정보가 잘못 들어온 경우 빈 데이터프레임을 반환 함
    if pd.isnull(twitterName) or twitterName == "":
        return tweets_df2;
        
    # Using TwitterSearchScraper to scrape data and append tweets to list
    for i,tweet in enumerate(sntwitter.TwitterSearchScraper('from:'+twitterName+' since:'+startDay+' until:'+endDay+'').get_items()):
        tweets_list1.append([tweet.url, tweet.date, tweet.id, tweet.content, tweet.username])
        #print(tweets_list1)
        # Creating a dataframe from the tweets list above 
        tweets_df2 = pd.DataFrame(tweets_list1, columns=['URL','Datetime', 'Tweet Id','Content', 'Username'])
    
    return tweets_df2;

 

• 리트윗 및 리플라이를 제거하는 함수 생성
  • 트위터 수집 후 해당 함수를 사용할 수 있음
  • 함수를 사용하는 경우 트윗 내용에서 리트윗/리플라이는 삭제된 후 남은 데이터를 반환
  • 입력 인자로 트윗 내용이 담긴 데이터프레임과 트윗 내용에 해당하는 컬럼 이름을 입력

# 인자 : 데이터 프레임 / 내용에 해당하는 컬럼의 이름
# 리턴되는 결과 : RT 및 리플라이를 제외한 결과 트윗 데이터 프레임
def remove_rt_reply(df,contentCol):
    # content 의 가장 앞에 '@' 라는 문자열이  있는 경우 = Reply 로 판단.
    # content 의 가장 앞에 'RT @' 라는 문자열이 있는 경우 = retweet 으로 판단.
    # 결과적으로 내용 앞에 @ 가 있는 경우를 제거함으로서 리플라이 및 리트윗을 제거하고 남은 데이터 프레잉을 리턴함
    rs = df.copy(deep=True)
    row = -1
    target = rs[contentCol]
    
    
    rs['retflag'] = False
    for i in target:
        row = row + 1
        
        if(i[0:1] == "@" or i[0:2] == "RT"):
            rs['retflag'][row] = True
        else:
            rs['retflag'][row] = False
        
    rs_L1 = rs[rs['retflag'] == False]
    
    del rs_L1['retflag']
    rs_L1 = rs_L1.reset_index(drop=True)
    
    return rs_L1;
        
        

 

• 특정 키워드를 추출하는 함수 생성
  • 수집된 트윗 내용으로부터 특정 키워드가 포함 되어있는지 찾아낸 후 키워드가 포함 되어있는 트윗을 찾아줌.
  • 키워드가 포함되어있지 않는 경우 이를 삭제할 것인지 여부를 입력 파라미터에서 결정 가능
  • 다수의 키워드를 한번에 입력할 수 있으며 이 경우 키워드 중 하나 이상이 트윗 내용에 존재하는 경우 참으로 간주됨

# 인자 : 데이터 프레임 / 키워드 리스트 / 내용에 해당하는 컬럼 이름 / 헤시태그만 찾을 것인지 여부 / 키워드에 해당하는 내용이 없는 트윗 삭제 여부
    # 리턴되는 결과 : 찾고자 하는 키워드가 있는 데이터가 존재하는 데이터 프레임
        # 조건 1: isOnlyHashtag 가 True 인 경우 키워드 앞에 # 를 붙여서 헤시태그에 해당하는 내용만 찾음 (False 인 경우 순수하게 키워드 존재 여부로 찾아주)
        # 조건 2 : isremove 가 True 인 경우 키워드를 찾지 못한 내용은 삭제한 후 리턴 (False 인경우 flag 만 붙여준 후 리턴)
def search_keyword(df,keyword,contentCol,isOnlyHashtag,isremove):
    
    rs = df.copy(deep=True)
    target = rs[contentCol]
    keyword_low = []
    # 오로지 헤시태그만 찾고자 하는 경우 키워드 앞에 # 을 붙이는 과정을 진행한다.
    if(isOnlyHashtag == True):
        for k in range(0,len(keyword),1):
            keyword[k] = '#' + keyword[k]
    else:
        keyword = keyword
        
    for k in range(0,len(keyword),1):
        keyword_low.append(keyword[k].lower())
            
    rs['findKeywordFlag'] = False
    rs['findKeyword'] = ''
    
    row = -1
    for i in target: # 콘텐츠의 내용
        i_low = i.lower()
        row = row + 1
        for k in keyword_low: # 키워드 (대소문자는 구분하지 않음)
            
            if(i_low.find(k) >= 0): 
                rs['findKeywordFlag'][row] = True
                key = rs['findKeyword'][row]
                rs['findKeyword'][row] = rs['findKeyword'][row] +  k + '|'
                
    if(isremove == True):
        rs_L1 = rs[rs['findKeywordFlag'] == True]
        rs_L1 = rs_L1.reset_index(drop=True)
    else:
        rs_L1 = rs
        
    return rs_L1;
    
    

 

[메인 프로그램 실행 코드]

• 메인 프로그램 실행
  • 검색할 시작날짜 / 끝 날짜에 해당하는 변수 지정
  • 찾아낼 키워드 지정 (복수 지정 가능)
  • 검색할 아이디 지정
  • 기타 로그 파일 위치 및 출력할 파일 위치 지정
  • 상기한 항목들을 모두선언 후 반복문을 통해 결과 파일을 출력한다.
  • 경우에 따라 특정 함수를 태우지 않고 단순 트윗 수집 이용 용도로만 사용도 가능
  • 기타 자세한 사항은 코드 부분의 주석 참고 바람

st_day = "2021-01-01" # 시작날짜 지정
ed_day = "2021-01-31" # 마지막 날짜 지정

my_keyword = ['마감'] # 찾고자 하는 키워드 지정

output_file_name = "./out/output.csv" # 출력할 파일 이름과 장소
log_file_path = "./log.txt" # 로그 파일의 이름과 장소

target_tweet_nmae = ['twittlions']

append_mode = False 

for sid in target_tweet_nmae:
    
    # 트위터 수집 관련 예외처리 구분
    try:
        if pd.isnull(sid) or sid == "":
            with open(log_file_path, "a") as file:
                file.write("잘못된 닉네임 입니다. 로그 저장 시각 : " + str(datetime.datetime.now()) + "\n")
            file.close()
            continue;
            
        result = read_tweet_list(sid,st_day,ed_day)
        print(len(result))
        with open(log_file_path, "a") as file:
            file.write(sid + " 아이디의 트위터 검색 완료! 총 " + str(len(result)) + "개의 트위터를 찾았습니다. 로그 저장 시각 : " + str(datetime.datetime.now()) + "\n")
        file.close()
    except:
        with open(log_file_path, "a") as file:
            file.write(sid + " 아이디의 트위터 검색 중 오류가 발생하였습니다. 해당 아이디를 건너뜁니다. 로그 저장 시각 : " + str(datetime.datetime.now()) + "\n")
        file.close()
        continue;


    # 리트윗 / 리플라이 제거
    # 인자 : 데이터 프레임 / 내용에 해당하는 컬럼 이름
    ## 리트윗을 삭제 하고자 하는 경우 아래의 코드 주석을 헤제하고 result_L2 변수 할당 하는 부분의 첫번재 인자를 result -> result_L1 로 변경하면 됨.
    # result_L1 = remove_rt_reply(result,'Content') 
    
    # 키워드가 존재하는 행 찾기
    # 인자 : 데이터 프레임 / 키워드 리스트 / 내용에 해당하는 컬럼 이름 / 헤시태그만 찾을 것인지 여부 / 키워드에 해당하는 내용이 없는 트윗 삭제 여부
    result_L2 = search_keyword(result,my_keyword,'Content',False,True)
    
    sleep(10) # 쿨타임을 줘서 과부하 방지
    
    # 데이터가 존재하는 경우 쓰기 (라인단위)
    if(len(result_L2) > 0):
                
        if append_mode == False:
            append_mode = True
            result_L2.to_csv(output_file_name + ".csv",index=False,header=True)
            
        elif append_mode == True:
            for i in range(len(result_L2)):
                result_L2.loc[[i]].to_csv(output_file_name,index=False,header=False,mode='a')
    
   

 

• 메인 프로그램 실행 예시
  • 상기 매인 코드에서 나타나듯이 수집 시작 날짜는 2021년 01월 01일 부터 01월 31일 까지 수집 대상 날짜로 지정하였음
  • 수집 대상 아이디는 삼성 라이온즈 아이디로 지정 하였음 (twittlions) 
  • 찾고자 하는 키워드는 '마감' 으로 지정하였음
  • 수집 이후 10초의 딜레이를 주었으며, 결과 출력의 경우 append 로 쌓임.

 

• 메인 프로그램 실행 결과

TEST.csv.csv

0.00MB

 

 참고 문헌

[논문]

• 없음

[보고서]

• 없음

[URL]

• 없음

 

 문의사항

[기상학/프로그래밍 언어]

• sangho.lee.1990@gmail.com

[해양학/천문학/빅데이터]

• saimang0804@gmail.com