def get_contributor_state(
  scoredNotes: pd.DataFrame,
  ratings: pd.DataFrame,
  noteStatusHistory: pd.DataFrame,
  userEnrollment: pd.DataFrame,
  log: bool = True,
) -> pd.DataFrame:
  """
  Given scored notes, ratings, note status history, the current user enrollment state, this
  uses the contributor counts over ratings and notes and transitions the user between the
  different enrollment states. If current user enrollment state is removed, do not change.

  Args:
      scoredNotes (pd.DataFrame): scored notes
      ratings (pd.DataFrame): all ratings
      noteStatusHistory (pd.DataFrame): history of note statuses
      userEnrollment (pd.DataFrame): User enrollment for BW participants.
      log (bool): Should we log
  Returns:
      pd.DataFrame: contributorScoresWithEnrollment The contributor scores with enrollments
  """
  with c.time_block("Contributor State: Setup"):
    # for users in state Earned Out Ack, update the timestamp of last earn out; this ensures they are only judged against
    # their rating target until they resume writing notes
    userEnrollment.loc[
      userEnrollment[c.enrollmentState] == c.earnedOutAcknowledged, c.timestampOfLastEarnOut
    ] = c.epochMillis

    # We need to consider only the last 5 notes for enrollment state. The ratings are aggregated historically.
    # For users who have earned out, we should only consider notes written since the earn out event
    scoredNotesWithLastEarnOut = scoredNotes.merge(
      userEnrollment[[c.participantIdKey, c.timestampOfLastEarnOut]],
      left_on=c.noteAuthorParticipantIdKey,
      right_on=c.participantIdKey,
      how="left",
      unsafeAllowed=c.timestampOfLastEarnOut,
    )
    # For users who don't appear in the userEnrollment file, set their timeStampOfLastEarnOut to default
    scoredNotesWithLastEarnOut[c.timestampOfLastEarnOut].fillna(1, inplace=True)

  with c.time_block("Contributor State: Contributor Scores"):
    contributorScores = get_contributor_scores(
      scoredNotesWithLastEarnOut,
      ratings,
      noteStatusHistory,
      lastNNotes=c.maxHistoryEarnOut,
      countNMRNotesLast=True,
      sinceLastEarnOut=True,
    )
    contributorScores.fillna(0, inplace=True)

  contributorScores[c.hasCrnhSinceEarnOut] = contributorScores[c.notesCurrentlyRatedNotHelpful] > 0

  with c.time_block("Contributor State: Top NH Tags Per Author"):
    # We merge in the top not helpful tags
    authorTopNotHelpfulTags = explanation_tags.get_top_nonhelpful_tags_per_author(
      noteStatusHistory, ratings
    )
    contributorScores = contributorScores.merge(
      authorTopNotHelpfulTags,
      left_on=c.raterParticipantIdKey,
      right_on=c.noteAuthorParticipantIdKey,
      how="outer",
      unsafeAllowed=c.hasCrnhSinceEarnOut,
    ).drop(columns=[c.noteAuthorParticipantIdKey])

  with c.time_block("Contributor State: Emerging Writers"):
    # We merge in the emerging writer data.
    emergingWriter = is_emerging_writer(scoredNotes)
    contributorScores = contributorScores.merge(
      emergingWriter,
      left_on=c.raterParticipantIdKey,
      right_on=c.noteAuthorParticipantIdKey,
      how="outer",
      unsafeAllowed=c.isEmergingWriterKey,
    ).drop(columns=[c.noteAuthorParticipantIdKey])

  with c.time_block("Contributor State: Combining"):
    # We merge the current enrollment state
    contributorScoresWithEnrollment = contributorScores.merge(
      userEnrollment,
      left_on=c.raterParticipantIdKey,
      right_on=c.participantIdKey,
      how="outer",
      unsafeAllowed={
        c.successfulRatingNeededToEarnIn,
        c.timestampOfLastStateChange,
        c.numberOfTimesEarnedOutKey,
        "coreBool",
        "expansionBool",
      },
    )

    # We set the new contributor state.
    contributorScoresWithEnrollment.fillna(
      inplace=True,
      value={
        c.successfulRatingNeededToEarnIn: c.ratingImpactForEarnIn,
        c.enrollmentState: c.newUser,
        c.isEmergingWriterKey: False,
      },
    )
    contributorScoresWithEnrollment[c.ratingImpact] = (
      contributorScoresWithEnrollment[c.successfulRatingTotal]
      - contributorScoresWithEnrollment[c.unsuccessfulRatingTotal]
      # 2x penalty for helpful ratings on CRNH notes
      - contributorScoresWithEnrollment[c.unsuccessfulRatingNotHelpfulCount]
    )

    contributorScoresWithEnrollment.loc[
      should_earn_in(contributorScoresWithEnrollment), c.enrollmentState
    ] = c.enrollmentStateToThrift[c.earnedIn]
    contributorScoresWithEnrollment.loc[
      should_earn_in(contributorScoresWithEnrollment), c.timestampOfLastStateChange
    ] = c.epochMillis

    contributorScoresWithEnrollment.loc[
      newly_at_risk(contributorScoresWithEnrollment), c.enrollmentState
    ] = c.enrollmentStateToThrift[c.atRisk]
    contributorScoresWithEnrollment.loc[
      newly_at_risk(contributorScoresWithEnrollment), c.timestampOfLastStateChange
    ] = c.epochMillis

    # for earned out users, first increment the number of times they have earned out,
    # use this to overwrite successful rating needed to earn in,
    # then set new state
    earnedOutUsers = is_earned_out(contributorScoresWithEnrollment)
    contributorScoresWithEnrollment.loc[earnedOutUsers, c.numberOfTimesEarnedOutKey] = (
      contributorScoresWithEnrollment.loc[earnedOutUsers, c.numberOfTimesEarnedOutKey] + 1
    )

    # use earned out no ack internally to identify newly earned out users
    contributorScoresWithEnrollment.loc[
      earnedOutUsers, c.enrollmentState
    ] = c.enrollmentStateToThrift[c.earnedOutNoAcknowledge]

    contributorScoresWithEnrollment.loc[
      earnedOutUsers, c.timestampOfLastStateChange
    ] = c.epochMillis

    # at risk users transitioning back to earned in
    contributorScoresWithEnrollment.loc[
      newly_earned_in(contributorScoresWithEnrollment), c.enrollmentState
    ] = c.enrollmentStateToThrift[c.earnedIn]
    contributorScoresWithEnrollment.loc[
      newly_earned_in(contributorScoresWithEnrollment), c.timestampOfLastStateChange
    ] = c.epochMillis

    contributorScoresWithEnrollment[c.enrollmentState] = contributorScoresWithEnrollment[
      c.enrollmentState
    ].map(_transform_to_thrift_code)

    mappedUserEnrollment = userEnrollment[
      [c.participantIdKey, c.timestampOfLastEarnOut, c.enrollmentState]
    ]
    mappedUserEnrollment[c.enrollmentState] = mappedUserEnrollment[c.enrollmentState].map(
      _transform_to_thrift_code
    )
    mappedUserEnrollment = mappedUserEnrollment.rename(
      columns={c.enrollmentState: c.enrollmentState + "_prev"}
    )

    # This addresses an issue in the TSV dump in HDFS getting corrupted. It removes lines
    # users that do not have an id.
    contributorScoresWithEnrollment.dropna(subset=[c.raterParticipantIdKey], inplace=True)

  if log:
    logger.info("Enrollment State")
    logger.info(
      f"Number of Earned In {len(contributorScoresWithEnrollment[contributorScoresWithEnrollment[c.enrollmentState] == 0])}"
    )
    logger.info(
      f"Number At Risk {len(contributorScoresWithEnrollment[contributorScoresWithEnrollment[c.enrollmentState] == 1])}"
    )
    logger.info(
      f"Number of Earn Out No Ack {len(contributorScoresWithEnrollment[contributorScoresWithEnrollment[c.enrollmentState] == 2])}"
    )
    logger.info(
      f"Number of Earned Out Ack {len(contributorScoresWithEnrollment[contributorScoresWithEnrollment[c.enrollmentState] == 3])}"
    )
    logger.info(
      f"Number of New Users {len(contributorScoresWithEnrollment[contributorScoresWithEnrollment[c.enrollmentState] == 4])}"
    )

  return contributorScoresWithEnrollment, mappedUserEnrollment

  Given scored notes, ratings, note status history, the current user enrollment state, this
  uses the contributor counts over ratings and notes and transitions the user between the
  different enrollment states. If current user enrollment state is removed, do not change.

  Args:
      scoredNotes (pd.DataFrame): scored notes
      ratings (pd.DataFrame): all ratings
      noteStatusHistory (pd.DataFrame): history of note statuses
      userEnrollment (pd.DataFrame): User enrollment for BW participants.
      log (bool): Should we log
  Returns:
      pd.DataFrame: contributorScoresWithEnrollment The contributor scores with enrollments
  
sourcecode/scoring/contributor_state.py