영화 데이터셋으로
Neo4j 그래프를 짓다

-- 각 인물이 고유한 이름을 가진다고 가정할 수 있는 경우.
-- 한 사람이 여러 역할(배우이자 감독)을 맡아도 단일 노드로 표현되도록 보장한다.
-- 중복 노드의 우발적 생성을 막아 그래프를 깨끗하고 효율적으로 유지한다.
CREATE CONSTRAINT unique_person_name IF NOT EXISTS
FOR (p:Person)
REQUIRE p.person_name IS UNIQUE;

-- 앞의 고유성 제약은 데이터에 대한 가정에 기초한다. 실세계에서는
-- 같은 이름의 다른 개인을 만나는 일이 흔하다. 이럴 때는 IMDb나 TMDb 같은
-- 외부 소스의 person_id처럼 더 신뢰할 수 있는 식별자로 고유성을 강제한다.
CREATE CONSTRAINT unique_person_id IF NOT EXISTS
FOR (p:Person)
REQUIRE p.person_id IS UNIQUE;

-- 고유성은 강제하지 않지만 이름으로 인물을 자주 검색한다면,
-- person_name 인덱스가 질의 성능을 크게 개선한다.
CREATE INDEX person_name_index IF NOT EXISTS
FOR (p:Person)
ON (p.person_name);

-- 영화는 특히 추천 시스템이나 검색 기능에서 제목으로 자주 질의된다.
-- title 인덱싱은 특정 영화 검색 시 빠른 조회를 보장한다.
CREATE INDEX movie_title_index IF NOT EXISTS
FOR (m:Movie)
ON (m.title);

-- 영화 속 특정 역할(주연, 카메오)로 배우를 걸러야 한다면,
-- ACTED_IN 관계의 role 인덱싱이 전체 관계 스캔을 피해 질의를 가속한다.
CREATE INDEX acted_in_role_index IF NOT EXISTS
FOR ()-[r:ACTED_IN]-()
ON (r.role);

# CSV 파일 로딩
df = pd.read_csv('./raw_data/credits.csv')

# 관련 출연진 정보를 추출하는 함수
def extract_cast(cast_str):
    cast_list = ast.literal_eval(cast_str)
    return [
        {'actor_id': c['id'], 'name': c['name'],
         'character': c['character'], 'cast_id': c['cast_id']}
        for c in cast_list
    ]

# 관련 제작진 정보를 추출하는 함수 — 감독과 프로듀서만
def extract_crew(crew_str):
    crew_list = ast.literal_eval(crew_str)
    relevant_jobs = ['Director', 'Producer']
    return [
        {'crew_id': c['id'], 'name': c['name'], 'job': c['job']}
        for c in crew_list if c['job'] in relevant_jobs
    ]

# 각 행에 추출 함수 적용
df['cast'] = df['cast'].apply(extract_cast)
df['crew'] = df['crew'].apply(extract_crew)

# 리스트를 개별 행으로 전개
df_cast = df.explode('cast').dropna(subset=['cast'])
df_crew = df.explode('crew').dropna(subset=['crew'])

# 전개된 데이터 정규화
df_cast_normalized = pd.json_normalize(df_cast['cast'])
df_crew_normalized = pd.json_normalize(df_crew['crew'])

# 인덱스 중복 방지를 위한 리셋, 중복 행 제거
df_cast_normalized = df_cast_normalized.reset_index(drop=True).drop_duplicates()
df_crew_normalized = df_crew_normalized.reset_index(drop=True).drop_duplicates()

# 정규화된 DataFrame에 영화 ID를 다시 붙인다
df_cast_normalized['tmdbId'] = df_cast.reset_index(drop=True)['id']
df_crew_normalized['tmdbId'] = df_crew.reset_index(drop=True)['id']

# 갱신된 컬럼명으로 저장
df_cast_normalized.to_csv(os.path.join(output_dir, 'normalized_cast.csv'), index=False)
df_crew_normalized.to_csv(os.path.join(output_dir, 'normalized_crew.csv'), index=False)

# 검증용 샘플 출력
print("Sample of normalized cast data:"); print(df_cast_normalized.head())
print("Sample of normalized crew data:"); print(df_crew_normalized.head())

# CSV 파일 로딩
df = pd.read_csv('./raw_data/keywords.csv')

# 키워드 추출·정규화 함수
def normalize_keywords(keyword_str):
    if pd.isna(keyword_str) or not isinstance(keyword_str, str):
        return []          # NaN이거나 문자열이 아니면 빈 리스트
    # 문자열화된 JSON을 딕셔너리 리스트로 변환
    keyword_list = ast.literal_eval(keyword_str)
    # 각 키워드의 'name'만 리스트로 반환
    return [kw['name'] for kw in keyword_list]

df['keywords'] = df['keywords'].apply(normalize_keywords)

# tmdbId별로 모든 키워드를 한 행으로 결합
df_keywords_aggregated = df.groupby('id', as_index=False).agg({
    'keywords': lambda x: ', '.join(sum(x, []))
})
df_keywords_aggregated.rename(columns={'id': 'tmdbId'}, inplace=True)
df_keywords_aggregated.to_csv(
    os.path.join(output_dir, 'normalized_keywords.csv'), index=False)
print(df_keywords_aggregated.head())

# 장르 추출·정규화 함수
def extract_genres(genres_str):
    if pd.isna(genres_str) or not isinstance(genres_str, str):
        return []
    genres_list = ast.literal_eval(genres_str)
    return [{'genre_id': int(g['id']), 'genre_name': g['name']}
            for g in genres_list]

# 제작사 추출·정규화 함수
def extract_production_companies(companies_str):
    if pd.isna(companies_str) or not isinstance(companies_str, str):
        return []
    companies_list = ast.literal_eval(companies_str)
    if isinstance(companies_list, list):
        return [{'company_id': int(c['id']),
                 'company_name': c['name']}
                for c in companies_list]
    return []

# "belongs_to_collection"에서 "name"만 추출
def extract_collection_name(collection_str):
    if isinstance(collection_str, str):
        try:
            collection_dict = ast.literal_eval(collection_str)
            if isinstance(collection_dict, dict):
                return collection_dict.get('name', "None")
        except (ValueError, SyntaxError):   # 문자열 파싱 실패 처리
            return "None"
    return "None"

df_movies = df[['id', 'original_title', 'adult', 'budget', 'imdb_id',
    'original_language', 'revenue', 'tagline', 'title',
    'release_date', 'runtime', 'overview', 'belongs_to_collection']].copy()
df_movies['belongs_to_collection'] = \
    df_movies['belongs_to_collection'].apply(extract_collection_name)
df_movies['adult'] = df_movies['adult'].apply(
    lambda x: 1 if x == 'TRUE' else 0)   # 'adult'를 정수로 변환
df_movies.rename(columns={'id': 'tmdbId'}, inplace=True)
df_movies.to_csv('./normalized_data/normalized_movies.csv', index=False)

"CREATE CONSTRAINT unique_tmdb_id IF NOT EXISTS FOR (m:Movie)
 REQUIRE m.tmdbId IS UNIQUE;",
"CREATE CONSTRAINT unique_movie_id IF NOT EXISTS FOR (m:Movie)
 REQUIRE m.movieId IS UNIQUE;",
"CREATE CONSTRAINT unique_prod_id IF NOT EXISTS FOR
 (p:ProductionCompany) REQUIRE p.company_id IS UNIQUE;",
"CREATE CONSTRAINT unique_genre_id IF NOT EXISTS FOR (g:Genre)
 REQUIRE g.genre_id IS UNIQUE;",
"CREATE CONSTRAINT unique_lang_id IF NOT EXISTS FOR
 (l:SpokenLanguage) REQUIRE l.language_code IS UNIQUE;",
"CREATE CONSTRAINT unique_country_id IF NOT EXISTS FOR (c:Country)
 REQUIRE c.country_code IS UNIQUE;",
"CREATE INDEX actor_id IF NOT EXISTS FOR (p:Person) ON (p.actor_id);",
"CREATE INDEX crew_id IF NOT EXISTS FOR (p:Person) ON (p.crew_id);",
"CREATE INDEX movieId IF NOT EXISTS FOR (m:Movie) ON (m.movieId);",
"CREATE INDEX user_id IF NOT EXISTS FOR (p:Person) ON (p.user_id);"

graph.load_movies(
    'https://storage.googleapis.com/movies-packt/normalized_movies.csv',
    movie_limit)

MATCH (m:Movie)-[:HAS_GENRE]->(g:Genre)
RETURN m.title, g.genre_name
LIMIT 10;

MATCH (a:Actor {name: 'Tom Hanks'})-[:ACTED_IN*1..3]-(m:Movie)
RETURN DISTINCT m.title;

MATCH path = (a:Actor)-[:ACTED_IN]->(m:Movie)
RETURN path;

MATCH (a:Actor {name: "Tom Hanks"})-[:ACTED_IN]->(m1:Movie)
      <-[:DIRECTED_BY]-(d:Director)
MATCH (a)-[:ACTED_IN]->(m2:Movie)<-[:DIRECTED_BY]-(d)
WHERE m1 <> m2
RETURN a.name AS actor, d.name AS director,
       collect(DISTINCT m1.title) + collect(DISTINCT m2.title) AS movies

CALL {
  MATCH (m:Movie)-[:HAS_GENRE]->(g:Genre {name: 'Action'})
  RETURN m
}
MATCH (m)-[:DIRECTED_BY]->(d:Director)
RETURN d.name, COUNT(m) AS action_movies_directed;

LOAD CSV WITH HEADERS FROM $csvFile AS row
CALL (row) {
  MATCH (m:Movie {movieId: toInteger(row.movieId)})
  WITH m, row
  MERGE (p:Person {user_id: toInteger(row.userId)})
  ON CREATE SET p.role = 'user'
  MERGE (p)-[r:RATED]->(m)
  ON CREATE SET r.rating = toFloat(row.rating),
                r.timestamp = toInteger(row.timestamp)
} IN TRANSACTIONS OF 50000 ROWS;

MATCH (m:Movie)
WHERE m.revenue > 100000000
CALL {
  WITH m
  MATCH (m)-[:HAS_GENRE]->(g:Genre)
  RETURN g.name AS genre
}
RETURN m.title, genre;